initial commit

Dockerfile

@@ -0,0 +1,25 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# Install dependencies
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy application code
+COPY . .
+
+# Environment variables
+ENV ENABLED_PROVIDERS="openfda,clinical_guidelines,cms_pricing"
+ENV LOG_LEVEL="INFO"
+ENV PYTHONUNBUFFERED=1
+
+# Expose port
+EXPOSE 7860
+
+# Health check
+HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
+  CMD python -c "import httpx; httpx.get('http://localhost:7860/health')"
+
+# Run HTTP streaming server
+CMD ["python", "server_http.py"]

README.md

@@ -1,12 +1,157 @@
----
-title: Healthcare Api Mcp
-emoji: 😻
-colorFrom: red
-colorTo: indigo
-sdk: docker
-pinned: false
-license: other
-short_description: multi healthcare api mcp wrapper
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Healthcare API MCP
+
+A Model Context Protocol (MCP) server providing access to public healthcare data APIs including FDA adverse events, clinical guidelines, and Medicare pricing information.
+
+## Features
+
+### OpenFDA Provider (14 tools)
+- **Drug Adverse Events**: Query FAERS database for medication safety information
+- **Drug Labels**: Search FDA-approved drug labeling information
+- **Drug Recalls**: Find drug recall and enforcement reports
+- **NDC Directory**: Look up National Drug Codes
+- **Drugs@FDA**: Search approved drug products
+- **Device Events & Recalls**: Medical device adverse events and recalls
+- **Device Classifications**: Device classification database
+- **510(k) & PMA**: Device clearances and approvals
+- **Food Recalls & Events**: Food safety information
+
+### Clinical Guidelines Provider (5 tools)
+- **Preventive Recommendations**: USPSTF evidence-based screening guidelines
+- **Vaccine Schedule**: CDC immunization recommendations
+- **Search Guidelines**: Find guidelines by condition or screening type
+- **Lifestyle Recommendations**: Evidence-based wellness guidance
+- **Care Plan Generator**: Comprehensive 6-month care planning
+
+### CMS Pricing Provider (5 tools)
+- **Medicare Fee Schedule**: Get reimbursement rates by HCPCS code
+- **Procedure Cost Estimates**: Typical cost ranges by procedure
+- **Search Procedure Codes**: Find HCPCS/CPT codes
+- **Facility Cost Comparison**: Compare prices across facilities
+- **Out-of-Pocket Estimates**: Calculate patient responsibility
+
+## Installation
+
+```bash
+pip install -r requirements.txt
+```
+
+## Configuration
+
+Set environment variables to configure the server:
+
+```bash
+# Enable specific providers (comma-separated)
+export ENABLED_PROVIDERS="openfda,clinical_guidelines,cms_pricing"
+
+# Optional: OpenFDA API key for higher rate limits
+export OPENFDA_API_KEY="your_api_key_here"
+
+# Logging level
+export LOG_LEVEL="INFO"
+```
+
+## Usage
+
+### Run as MCP Server (stdio)
+
+```bash
+python -m healthcare_api_mcp
+```
+
+### Use with Claude Desktop
+
+Add to your Claude Desktop config (`claude_desktop_config.json`):
+
+```json
+{
+  "mcpServers": {
+    "healthcare-api": {
+      "command": "python",
+      "args": ["-m", "healthcare_api_mcp"],
+      "env": {
+        "ENABLED_PROVIDERS": "openfda,clinical_guidelines,cms_pricing"
+      }
+    }
+  }
+}
+```
+
+### Filter Providers
+
+Run only specific providers:
+
+```bash
+# Only FDA data
+ENABLED_PROVIDERS=openfda python -m healthcare_api_mcp
+
+# Only clinical guidelines
+ENABLED_PROVIDERS=clinical_guidelines python -m healthcare_api_mcp
+
+# FDA and CMS pricing
+ENABLED_PROVIDERS=openfda,cms_pricing python -m healthcare_api_mcp
+```
+
+## Tool Naming
+
+All tools are namespaced by provider:
+- OpenFDA: `openfda_*` (e.g., `openfda_search_adverse_events`)
+- Clinical Guidelines: `clinical_*` (e.g., `clinical_get_vaccine_schedule`)
+- CMS Pricing: `cms_*` (e.g., `cms_get_medicare_fee`)
+
+## Example Queries
+
+### Check Drug Safety
+```
+Agent: "Check if Simvastatin has any FDA recalls"
+→ Uses: openfda_search_drug_recalls
+```
+
+### Get Preventive Care Recommendations
+```
+Agent: "What screenings does a 55-year-old female need?"
+→ Uses: clinical_get_preventive_recommendations
+```
+
+### Estimate Procedure Cost
+```
+Agent: "How much does a colonoscopy cost with Medicare?"
+→ Uses: cms_estimate_procedure_cost, cms_estimate_out_of_pocket
+```
+
+## Data Sources
+
+- **OpenFDA**: Real-time access to FDA's public APIs
+- **Clinical Guidelines**: Evidence-based recommendations from USPSTF, CDC, AHRQ
+- **CMS Pricing**: Sample data based on Medicare fee schedules (2024)
+
+## Architecture
+
+```
+healthcare_api_mcp/
+├── server.py                    # Main MCP server with stdio
+├── config.py                    # Configuration management
+├── core/
+│   ├── base_provider.py         # Abstract provider base class
+│   ├── decorators.py            # @safe_json_return, @with_retry
+│   └── http_client.py           # Shared HTTP client factory
+└── providers/
+    ├── openfda_provider.py      # FDA APIs
+    ├── clinical_guidelines_provider.py
+    └── cms_pricing_provider.py
+```
+
+## Benefits
+
+✅ **Zero duplication**: Shared utilities across all providers
+✅ **Proper JSON returns**: All tools return JSON strings (no dict inconsistencies)
+✅ **Modular**: Enable only the providers you need
+✅ **Agent-friendly**: Standard MCP stdio protocol
+✅ **Production-ready**: Error handling, retry logic, logging
+
+## License
+
+Copyright (c) 2025. All Rights Reserved.
+
+## HuggingFace MCP 1st Birthday Hackathon
+
+This project is submitted for Track 1 of the HuggingFace MCP 1st Birthday Hackathon.

__init__.py

@@ -0,0 +1,3 @@
+"""Healthcare API MCP - Public health data APIs."""
+
+__version__ = "1.0.0"

__main__.py

@@ -0,0 +1,7 @@
+"""Entry point for running the MCP server as a module."""
+
+from .server import main
+import asyncio
+
+if __name__ == "__main__":
+    asyncio.run(main())

config.py

@@ -0,0 +1,44 @@
+"""Configuration for Healthcare API MCP."""
+
+import os
+from typing import List, Optional
+from pydantic import BaseModel, Field
+
+
+class ProviderConfig(BaseModel):
+    """Configuration for a provider."""
+    name: str
+    enabled: bool = True
+    api_key: Optional[str] = None
+
+
+class MCPConfig(BaseModel):
+    """Main MCP server configuration."""
+
+    # Enabled providers (comma-separated or list)
+    enabled_providers: List[str] = Field(
+        default_factory=lambda: ["openfda", "clinical_guidelines", "cms_pricing"]
+    )
+
+    # OpenFDA API key (optional but recommended for higher rate limits)
+    openfda_api_key: Optional[str] = Field(default=None)
+
+    # Logging level
+    log_level: str = Field(default="INFO")
+
+    @classmethod
+    def from_env(cls) -> "MCPConfig":
+        """Load configuration from environment variables."""
+
+        # Parse enabled providers
+        enabled_str = os.getenv("ENABLED_PROVIDERS", "")
+        if enabled_str:
+            enabled_providers = [p.strip() for p in enabled_str.split(",")]
+        else:
+            enabled_providers = ["openfda", "clinical_guidelines", "cms_pricing"]
+
+        return cls(
+            enabled_providers=enabled_providers,
+            openfda_api_key=os.getenv("OPENFDA_API_KEY"),
+            log_level=os.getenv("LOG_LEVEL", "INFO")
+        )

core/__init__.py

@@ -0,0 +1,7 @@
+"""Core utilities for Healthcare API MCP."""
+
+from .base_provider import BaseProvider
+from .decorators import safe_json_return, with_retry
+from .http_client import create_http_client
+
+__all__ = ["BaseProvider", "safe_json_return", "with_retry", "create_http_client"]

core/base_provider.py

@@ -0,0 +1,58 @@
+"""Base provider class for healthcare APIs."""
+
+from abc import ABC, abstractmethod
+from typing import List, Callable
+import httpx
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class BaseProvider(ABC):
+    """Abstract base class for healthcare API providers."""
+
+    def __init__(self, name: str, client: httpx.AsyncClient):
+        """
+        Initialize provider.
+
+        Args:
+            name: Provider name (used for tool prefixing)
+            client: Async HTTP client
+        """
+        self.name = name
+        self.client = client
+        self._tools = []
+
+    @abstractmethod
+    async def initialize(self) -> None:
+        """
+        Initialize provider (e.g., test API connection).
+        Override in subclasses if needed.
+        """
+        pass
+
+    @abstractmethod
+    def get_tools(self) -> List[Callable]:
+        """
+        Return list of MCP tool functions provided by this provider.
+
+        Returns:
+            List of tool functions
+        """
+        pass
+
+    async def cleanup(self) -> None:
+        """
+        Cleanup provider resources.
+        Override in subclasses if needed.
+        """
+        pass
+
+    def register_tool(self, func: Callable) -> None:
+        """Register a tool function."""
+        self._tools.append(func)
+
+    @property
+    def tools(self) -> List[Callable]:
+        """Get all registered tools."""
+        return self._tools

core/decorators.py

@@ -0,0 +1,85 @@
+"""Shared decorators for MCP tools."""
+
+import functools
+import json
+import logging
+from typing import Any, Callable, TypeVar
+from tenacity import retry, stop_after_attempt, wait_exponential
+
+logger = logging.getLogger(__name__)
+
+F = TypeVar('F', bound=Callable[..., Any])
+
+
+def safe_json_return(func: F) -> F:
+    """
+    Decorator that ensures MCP tools always return JSON strings.
+    Handles dicts, lists, None, and exceptions consistently.
+    """
+    @functools.wraps(func)
+    async def async_wrapper(*args, **kwargs):
+        try:
+            result = await func(*args, **kwargs)
+
+            # Handle None
+            if result is None:
+                return json.dumps({"result": None})
+
+            # Handle dict or list
+            if isinstance(result, (dict, list)):
+                return json.dumps(result, ensure_ascii=False)
+
+            # Handle string (might already be JSON)
+            if isinstance(result, str):
+                return result
+
+            # Handle other types
+            return json.dumps({"result": str(result)})
+
+        except Exception as e:
+            logger.exception(f"Error in {func.__name__}")
+            return json.dumps({
+                "error": str(e),
+                "error_type": type(e).__name__
+            })
+
+    @functools.wraps(func)
+    def sync_wrapper(*args, **kwargs):
+        try:
+            result = func(*args, **kwargs)
+
+            if result is None:
+                return json.dumps({"result": None})
+
+            if isinstance(result, (dict, list)):
+                return json.dumps(result, ensure_ascii=False)
+
+            if isinstance(result, str):
+                return result
+
+            return json.dumps({"result": str(result)})
+
+        except Exception as e:
+            logger.exception(f"Error in {func.__name__}")
+            return json.dumps({
+                "error": str(e),
+                "error_type": type(e).__name__
+            })
+
+    # Return appropriate wrapper based on function type
+    import inspect
+    if inspect.iscoroutinefunction(func):
+        return async_wrapper
+    else:
+        return sync_wrapper
+
+
+def with_retry(func: F) -> F:
+    """
+    Decorator for retrying failed HTTP requests with exponential backoff.
+    """
+    return retry(
+        stop=stop_after_attempt(3),
+        wait=wait_exponential(multiplier=1, min=2, max=8),
+        reraise=True
+    )(func)

core/http_client.py

@@ -0,0 +1,35 @@
+"""Shared HTTP client factory."""
+
+import httpx
+from typing import Optional, Dict
+
+
+def create_http_client(
+    timeout: float = 30.0,
+    headers: Optional[Dict[str, str]] = None,
+    follow_redirects: bool = True
+) -> httpx.AsyncClient:
+    """
+    Create a configured async HTTP client.
+
+    Args:
+        timeout: Request timeout in seconds
+        headers: Default headers to include
+        follow_redirects: Whether to follow redirects
+
+    Returns:
+        Configured httpx.AsyncClient
+    """
+    default_headers = {
+        "Accept": "application/json",
+        "Content-Type": "application/json",
+    }
+
+    if headers:
+        default_headers.update(headers)
+
+    return httpx.AsyncClient(
+        timeout=timeout,
+        headers=default_headers,
+        follow_redirects=follow_redirects
+    )

providers/__init__.py

@@ -0,0 +1,20 @@
+"""Healthcare API providers."""
+
+from typing import Dict, Type
+from ..core.base_provider import BaseProvider
+
+# Provider registry
+PROVIDER_REGISTRY: Dict[str, Type[BaseProvider]] = {}
+
+
+def register_provider(name: str):
+    """Decorator to register a provider."""
+    def decorator(cls: Type[BaseProvider]):
+        PROVIDER_REGISTRY[name] = cls
+        return cls
+    return decorator
+
+
+def get_provider_class(name: str) -> Type[BaseProvider]:
+    """Get provider class by name."""
+    return PROVIDER_REGISTRY.get(name)

providers/clinical_guidelines_provider.py

@@ -0,0 +1,512 @@
+"""Clinical Guidelines Provider for USPSTF, CDC, and AHRQ guidelines."""
+
+import logging
+from typing import List, Dict, Any, Callable
+import httpx
+
+from ..core.base_provider import BaseProvider
+from ..core.decorators import safe_json_return
+from . import register_provider
+
+logger = logging.getLogger(__name__)
+
+# Evidence-based preventive care guidelines
+PREVENTIVE_GUIDELINES = {
+    "diabetes_screening": [
+        {
+            "title": "Type 2 Diabetes Screening",
+            "recommendation": "Screen for prediabetes and type 2 diabetes in adults aged 35 to 70 years who have overweight or obesity",
+            "grade": "B",
+            "age_range": "35-70 years",
+            "frequency": "Every 3 years if normal",
+            "source": "USPSTF",
+            "conditions": ["overweight", "obesity"],
+            "citation": "USPSTF 2021"
+        }
+    ],
+    "colorectal_screening": [
+        {
+            "title": "Colorectal Cancer Screening",
+            "recommendation": "Screen for colorectal cancer starting at age 45 years",
+            "grade": "A",
+            "age_range": "45-75 years",
+            "frequency": "Every 10 years (colonoscopy) or annually (FIT)",
+            "source": "USPSTF",
+            "methods": ["Colonoscopy every 10 years", "FIT annually", "CT colonography every 5 years"],
+            "citation": "USPSTF 2021"
+        }
+    ],
+    "hypertension_screening": [
+        {
+            "title": "Hypertension Screening",
+            "recommendation": "Screen for high blood pressure in adults 18 years or older",
+            "grade": "A",
+            "age_range": "18+ years",
+            "frequency": "Annually for adults 40+ or at increased risk; every 3-5 years for adults 18-39 with normal BP",
+            "source": "USPSTF",
+            "citation": "USPSTF 2021"
+        }
+    ],
+    "cholesterol_screening": [
+        {
+            "title": "Statin Use for Primary Prevention of CVD",
+            "recommendation": "Screen for cardiovascular disease risk factors beginning at age 40",
+            "grade": "B",
+            "age_range": "40-75 years",
+            "frequency": "Every 4-6 years",
+            "source": "USPSTF",
+            "conditions": ["1+ CVD risk factors"],
+            "citation": "USPSTF 2016"
+        }
+    ]
+}
+
+# CDC vaccine schedule recommendations
+VACCINE_SCHEDULE = {
+    "influenza": {
+        "vaccine": "Influenza (Flu)",
+        "recommendation": "Annual vaccination for all persons aged 6 months and older",
+        "age_groups": ["6 months+"],
+        "frequency": "Annually",
+        "source": "CDC",
+        "priority_groups": ["65+", "Pregnant", "Chronic conditions"]
+    },
+    "covid19": {
+        "vaccine": "COVID-19",
+        "recommendation": "Vaccination recommended for all persons 6 months and older",
+        "age_groups": ["6 months+"],
+        "frequency": "Updated annually",
+        "source": "CDC",
+        "priority_groups": ["65+", "Immunocompromised", "Healthcare workers"]
+    },
+    "pneumococcal": {
+        "vaccine": "Pneumococcal (PCV/PPSV)",
+        "recommendation": "Vaccination for adults 65 years or older and younger adults with certain conditions",
+        "age_groups": ["65+", "19-64 with risk factors"],
+        "frequency": "One-time or based on risk",
+        "source": "CDC",
+        "conditions": ["Chronic heart/lung disease", "Diabetes", "Immunocompromised"]
+    },
+    "shingles": {
+        "vaccine": "Shingles (Zoster)",
+        "recommendation": "2-dose series for adults 50 years and older",
+        "age_groups": ["50+"],
+        "frequency": "2 doses, 2-6 months apart",
+        "source": "CDC"
+    },
+    "tdap": {
+        "vaccine": "Tdap (Tetanus, Diphtheria, Pertussis)",
+        "recommendation": "One dose of Tdap, then Td booster every 10 years",
+        "age_groups": ["All adults"],
+        "frequency": "Every 10 years",
+        "source": "CDC"
+    },
+    "hepatitis_b": {
+        "vaccine": "Hepatitis B",
+        "recommendation": "Vaccination for adults at increased risk",
+        "age_groups": ["19-59 years (universal)", "60+ with risk factors"],
+        "frequency": "2 or 3-dose series",
+        "source": "CDC",
+        "risk_factors": ["Healthcare workers", "Diabetes", "Chronic liver disease"]
+    }
+}
+
+
+@register_provider("clinical_guidelines")
+class ClinicalGuidelinesProvider(BaseProvider):
+    """Provider for clinical guidelines and preventive care recommendations."""
+
+    def __init__(self, client: httpx.AsyncClient):
+        super().__init__("clinical_guidelines", client)
+
+    async def initialize(self) -> None:
+        """Initialize clinical guidelines provider."""
+        logger.info("Clinical Guidelines provider initialized")
+
+    def get_tools(self) -> List[Callable]:
+        """Return all clinical guideline tools."""
+        return [
+            self.clinical_get_preventive_recommendations,
+            self.clinical_get_vaccine_schedule,
+            self.clinical_search_guidelines,
+            self.clinical_get_lifestyle_recommendations,
+            self.clinical_generate_care_plan,
+        ]
+
+    @safe_json_return
+    async def clinical_get_preventive_recommendations(
+        self,
+        age: int,
+        sex: str,
+        conditions: List[str] = None
+    ) -> Dict[str, Any]:
+        """
+        Get personalized preventive care recommendations based on USPSTF guidelines.
+
+        Args:
+            age: Patient age in years
+            sex: Patient sex (male/female)
+            conditions: List of existing conditions (e.g., ["diabetes", "overweight"])
+
+        Returns:
+            Evidence-based screening and prevention recommendations with grades
+        """
+        if conditions is None:
+            conditions = []
+
+        recommendations = []
+        condition_lower = [c.lower() for c in conditions]
+
+        # Diabetes screening
+        if 35 <= age <= 70 and any(c in condition_lower for c in ["overweight", "obesity", "prediabetes"]):
+            recommendations.append({
+                **PREVENTIVE_GUIDELINES["diabetes_screening"][0],
+                "applicable": True,
+                "rationale": "Age and BMI criteria met"
+            })
+
+        # Colorectal cancer screening
+        if 45 <= age <= 75:
+            recommendations.append({
+                **PREVENTIVE_GUIDELINES["colorectal_screening"][0],
+                "applicable": True,
+                "rationale": "Age-based screening recommendation"
+            })
+
+        # Hypertension screening
+        if age >= 18:
+            freq = "Annually" if age >= 40 else "Every 3-5 years"
+            rec = PREVENTIVE_GUIDELINES["hypertension_screening"][0].copy()
+            rec["frequency"] = freq
+            rec["applicable"] = True
+            rec["rationale"] = "Universal screening for adults"
+            recommendations.append(rec)
+
+        # Cholesterol/CVD screening
+        if 40 <= age <= 75:
+            recommendations.append({
+                **PREVENTIVE_GUIDELINES["cholesterol_screening"][0],
+                "applicable": True,
+                "rationale": "Age-based cardiovascular risk screening"
+            })
+
+        # Sex-specific screenings
+        if sex.lower() == "female":
+            if 40 <= age <= 74:
+                recommendations.append({
+                    "title": "Breast Cancer Screening",
+                    "recommendation": "Biennial screening mammography for women aged 40-74",
+                    "grade": "B",
+                    "age_range": "40-74 years",
+                    "frequency": "Every 2 years",
+                    "source": "USPSTF",
+                    "applicable": True,
+                    "citation": "USPSTF 2024"
+                })
+            if 21 <= age <= 65:
+                recommendations.append({
+                    "title": "Cervical Cancer Screening",
+                    "recommendation": "Screening every 3 years with cytology (21-29) or every 5 years with HPV testing (30-65)",
+                    "grade": "A",
+                    "age_range": "21-65 years",
+                    "frequency": "Every 3-5 years based on method",
+                    "source": "USPSTF",
+                    "applicable": True,
+                    "citation": "USPSTF 2018"
+                })
+        elif sex.lower() == "male":
+            if 55 <= age <= 80 and "smoking history" in condition_lower:
+                recommendations.append({
+                    "title": "Lung Cancer Screening",
+                    "recommendation": "Annual screening with low-dose CT for adults 50-80 with 20 pack-year smoking history",
+                    "grade": "B",
+                    "age_range": "50-80 years",
+                    "frequency": "Annually",
+                    "source": "USPSTF",
+                    "applicable": True,
+                    "conditions": ["20 pack-year smoking history"],
+                    "citation": "USPSTF 2021"
+                })
+
+        return {
+            "patient_age": age,
+            "patient_sex": sex,
+            "conditions": conditions,
+            "recommendations": recommendations,
+            "total_recommendations": len(recommendations)
+        }
+
+    @safe_json_return
+    async def clinical_get_vaccine_schedule(
+        self,
+        age: int,
+        conditions: List[str] = None
+    ) -> Dict[str, Any]:
+        """
+        Get personalized vaccine recommendations based on CDC guidelines.
+
+        Args:
+            age: Patient age in years
+            conditions: List of health conditions
+
+        Returns:
+            Recommended vaccines with timing and priority
+        """
+        if conditions is None:
+            conditions = []
+
+        recommended_vaccines = []
+        condition_lower = [c.lower() for c in conditions]
+
+        # Universal vaccines
+        recommended_vaccines.append({
+            **VACCINE_SCHEDULE["influenza"],
+            "due": "Annually, preferably before flu season (Sept-Oct)",
+            "priority": "High" if age >= 65 or len(conditions) > 0 else "Standard"
+        })
+
+        recommended_vaccines.append({
+            **VACCINE_SCHEDULE["covid19"],
+            "due": "Check for updated formulation annually",
+            "priority": "High" if age >= 65 else "Standard"
+        })
+
+        # Age-specific vaccines
+        if age >= 65:
+            recommended_vaccines.append({
+                **VACCINE_SCHEDULE["pneumococcal"],
+                "due": "If not previously vaccinated",
+                "priority": "High"
+            })
+
+        if age >= 50:
+            recommended_vaccines.append({
+                **VACCINE_SCHEDULE["shingles"],
+                "due": "2-dose series if not previously vaccinated",
+                "priority": "High" if age >= 60 else "Standard"
+            })
+
+        # Condition-specific
+        if any(c in condition_lower for c in ["diabetes", "heart disease", "lung disease", "immunocompromised"]):
+            if age >= 19 and age < 65:
+                recommended_vaccines.append({
+                    **VACCINE_SCHEDULE["pneumococcal"],
+                    "due": "Based on risk factors",
+                    "priority": "High",
+                    "rationale": "Chronic medical condition"
+                })
+
+        # Tdap (universal adult)
+        recommended_vaccines.append({
+            **VACCINE_SCHEDULE["tdap"],
+            "due": "If not received in last 10 years",
+            "priority": "Standard"
+        })
+
+        # Hepatitis B
+        if 19 <= age <= 59:
+            recommended_vaccines.append({
+                **VACCINE_SCHEDULE["hepatitis_b"],
+                "due": "If not previously vaccinated",
+                "priority": "Standard"
+            })
+
+        return {
+            "patient_age": age,
+            "conditions": conditions,
+            "recommended_vaccines": recommended_vaccines,
+            "total_vaccines": len(recommended_vaccines)
+        }
+
+    @safe_json_return
+    async def clinical_search_guidelines(
+        self,
+        query: str,
+        source: str = "All"
+    ) -> Dict[str, Any]:
+        """
+        Search clinical guidelines by condition or screening type.
+
+        Args:
+            query: Search term (condition, screening type, etc.)
+            source: Filter by source (USPSTF, CDC, AHRQ, All)
+
+        Returns:
+            Matching guidelines with recommendations
+        """
+        query_lower = query.lower()
+        results = []
+
+        # Search preventive guidelines
+        for key, guidelines in PREVENTIVE_GUIDELINES.items():
+            if query_lower in key:
+                results.extend(guidelines)
+            else:
+                for g in guidelines:
+                    if query_lower in g.get("title", "").lower():
+                        results.append(g)
+
+        # Search vaccines
+        for key, vaccine in VACCINE_SCHEDULE.items():
+            if query_lower in key or query_lower in vaccine.get("vaccine", "").lower():
+                results.append({
+                    "title": vaccine["vaccine"],
+                    "recommendation": vaccine["recommendation"],
+                    "source": vaccine["source"],
+                    "type": "Immunization",
+                    "age_groups": vaccine.get("age_groups", []),
+                    "frequency": vaccine.get("frequency", "")
+                })
+
+        # Filter by source if specified
+        if source != "All":
+            results = [r for r in results if r.get("source", "").upper() == source.upper()]
+
+        return {
+            "query": query,
+            "source_filter": source,
+            "results": results,
+            "total": len(results)
+        }
+
+    @safe_json_return
+    async def clinical_get_lifestyle_recommendations(
+        self,
+        conditions: List[str],
+        age: int
+    ) -> Dict[str, Any]:
+        """
+        Get evidence-based lifestyle and wellness recommendations.
+
+        Args:
+            conditions: List of health conditions
+            age: Patient age
+
+        Returns:
+            Lifestyle modification recommendations
+        """
+        if conditions is None:
+            conditions = []
+
+        recommendations = []
+        condition_lower = [c.lower() for c in conditions]
+
+        # Universal recommendations
+        recommendations.append({
+            "category": "Physical Activity",
+            "recommendation": "At least 150 minutes of moderate-intensity aerobic activity per week",
+            "source": "CDC Physical Activity Guidelines",
+            "evidence_grade": "A"
+        })
+
+        recommendations.append({
+            "category": "Nutrition",
+            "recommendation": "Mediterranean-style diet rich in fruits, vegetables, whole grains, and healthy fats",
+            "source": "AHA Dietary Guidelines",
+            "evidence_grade": "A"
+        })
+
+        recommendations.append({
+            "category": "Sleep",
+            "recommendation": "7-9 hours of sleep per night for adults",
+            "source": "CDC Sleep Guidelines",
+            "evidence_grade": "A"
+        })
+
+        # Condition-specific
+        if "diabetes" in condition_lower:
+            recommendations.append({
+                "category": "Diabetes Management",
+                "recommendation": "Monitor blood glucose regularly, aim for HbA1c <7% for most adults",
+                "source": "ADA Standards of Care",
+                "evidence_grade": "A",
+                "specifics": ["Carbohydrate counting", "Regular foot examinations", "Annual eye exams"]
+            })
+
+        if "hypertension" in condition_lower:
+            recommendations.append({
+                "category": "Blood Pressure Management",
+                "recommendation": "DASH diet, sodium restriction (<2300mg/day), regular BP monitoring",
+                "source": "ACC/AHA Hypertension Guidelines",
+                "evidence_grade": "A",
+                "target": "BP <130/80 mmHg"
+            })
+
+        if any("heart" in c or "cardiac" in c or "cardiovascular" in c for c in condition_lower):
+            recommendations.append({
+                "category": "Cardiovascular Health",
+                "recommendation": "Cardiac rehabilitation if eligible, cholesterol management, smoking cessation",
+                "source": "AHA/ACC Guidelines",
+                "evidence_grade": "A"
+            })
+
+        if "obesity" in condition_lower or "overweight" in condition_lower:
+            recommendations.append({
+                "category": "Weight Management",
+                "recommendation": "5-10% weight loss through caloric reduction and increased physical activity",
+                "source": "USPSTF Obesity Guidelines",
+                "evidence_grade": "B"
+            })
+
+        return {
+            "conditions": conditions,
+            "patient_age": age,
+            "recommendations": recommendations,
+            "total": len(recommendations)
+        }
+
+    @safe_json_return
+    async def clinical_generate_care_plan(
+        self,
+        age: int,
+        sex: str,
+        conditions: List[str] = None,
+        current_medications: List[str] = None
+    ) -> Dict[str, Any]:
+        """
+        Generate comprehensive 6-month care plan with screening, vaccines, and lifestyle.
+
+        Args:
+            age: Patient age
+            sex: Patient sex
+            conditions: Existing conditions
+            current_medications: Current medication list
+
+        Returns:
+            Structured 6-month care plan
+        """
+        if conditions is None:
+            conditions = []
+        if current_medications is None:
+            current_medications = []
+
+        # Get all recommendation types
+        preventive = await self.clinical_get_preventive_recommendations(age, sex, conditions)
+        vaccines = await self.clinical_get_vaccine_schedule(age, conditions)
+        lifestyle = await self.clinical_get_lifestyle_recommendations(conditions, age)
+
+        # Build timeline
+        care_plan = {
+            "patient_profile": {
+                "age": age,
+                "sex": sex,
+                "conditions": conditions,
+                "current_medications": current_medications
+            },
+            "preventive_screenings": preventive.get("recommendations", []),
+            "immunizations": vaccines.get("recommended_vaccines", []),
+            "lifestyle_modifications": lifestyle.get("recommendations", []),
+            "timeline": {
+                "month_1": ["Schedule overdue screenings", "Update immunizations"],
+                "month_2": ["Follow up on screening results", "Begin lifestyle modifications"],
+                "month_3": ["Medication review with provider", "Assess lifestyle progress"],
+                "month_6": ["Comprehensive health review", "Update care plan"]
+            },
+            "follow_up_schedule": {
+                "primary_care": "Every 3-6 months" if conditions else "Annually",
+                "specialist": "As indicated by conditions" if conditions else "Not required",
+                "lab_work": "Per screening recommendations"
+            }
+        }
+
+        return care_plan

providers/cms_pricing_provider.py

@@ -0,0 +1,420 @@
+"""CMS Pricing Provider for Medicare fee schedules and cost transparency."""
+
+import logging
+from typing import List, Dict, Any, Callable
+import httpx
+
+from ..core.base_provider import BaseProvider
+from ..core.decorators import safe_json_return
+from . import register_provider
+
+logger = logging.getLogger(__name__)
+
+# Sample Medicare fee schedule data
+MEDICARE_FEE_SCHEDULE = {
+    "99213": {
+        "description": "Office visit, established patient, level 3",
+        "medicare_payment": 93.47,
+        "facility_rate": 93.47,
+        "non_facility_rate": 131.20,
+        "rvu": 1.92
+    },
+    "99214": {
+        "description": "Office visit, established patient, level 4",
+        "medicare_payment": 131.20,
+        "facility_rate": 131.20,
+        "non_facility_rate": 183.19,
+        "rvu": 2.80
+    },
+    "43239": {
+        "description": "Upper GI endoscopy with biopsy",
+        "medicare_payment": 248.54,
+        "facility_rate": 248.54,
+        "non_facility_rate": 450.00,
+        "rvu": 5.28
+    },
+    "45378": {
+        "description": "Colonoscopy, diagnostic",
+        "medicare_payment": 365.00,
+        "facility_rate": 365.00,
+        "non_facility_rate": 650.00,
+        "rvu": 7.75
+    },
+    "70450": {
+        "description": "CT scan, head without contrast",
+        "medicare_payment": 178.00,
+        "facility_rate": 178.00,
+        "non_facility_rate": 320.00,
+        "rvu": 3.78
+    },
+    "70553": {
+        "description": "MRI brain with contrast",
+        "medicare_payment": 485.00,
+        "facility_rate": 485.00,
+        "non_facility_rate": 875.00,
+        "rvu": 10.30
+    },
+    "80053": {
+        "description": "Comprehensive metabolic panel",
+        "medicare_payment": 14.00,
+        "facility_rate": 14.00,
+        "non_facility_rate": 18.00,
+        "rvu": 0.30
+    },
+    "85025": {
+        "description": "Complete blood count with differential",
+        "medicare_payment": 11.00,
+        "facility_rate": 11.00,
+        "non_facility_rate": 14.00,
+        "rvu": 0.23
+    },
+    "71046": {
+        "description": "Chest X-ray, 2 views",
+        "medicare_payment": 38.00,
+        "facility_rate": 38.00,
+        "non_facility_rate": 62.00,
+        "rvu": 0.81
+    }
+}
+
+# Sample procedure cost estimates
+PROCEDURE_COST_ESTIMATES = {
+    "colonoscopy": {
+        "procedure": "Diagnostic Colonoscopy",
+        "hcpcs_code": "45378",
+        "average_cost": 2750,
+        "low_estimate": 1500,
+        "high_estimate": 4500,
+        "medicare_rate": 365,
+        "factors": ["Facility type", "Anesthesia", "Biopsy performed", "Location"]
+    },
+    "mri": {
+        "procedure": "MRI Scan",
+        "hcpcs_code": "70553",
+        "average_cost": 1420,
+        "low_estimate": 400,
+        "high_estimate": 3500,
+        "medicare_rate": 485,
+        "factors": ["Body part", "With/without contrast", "Facility type"]
+    },
+    "ct scan": {
+        "procedure": "CT Scan",
+        "hcpcs_code": "70450",
+        "average_cost": 820,
+        "low_estimate": 300,
+        "high_estimate": 1800,
+        "medicare_rate": 178,
+        "factors": ["Body part", "With/without contrast", "Emergency vs scheduled"]
+    },
+    "blood test": {
+        "procedure": "Comprehensive Metabolic Panel",
+        "hcpcs_code": "80053",
+        "average_cost": 45,
+        "low_estimate": 15,
+        "high_estimate": 120,
+        "medicare_rate": 14,
+        "factors": ["Lab facility", "Stat vs routine", "Insurance negotiated rate"]
+    },
+    "x-ray": {
+        "procedure": "Chest X-Ray",
+        "hcpcs_code": "71046",
+        "average_cost": 125,
+        "low_estimate": 50,
+        "high_estimate": 300,
+        "medicare_rate": 38,
+        "factors": ["Number of views", "Facility type", "Interpretation included"]
+    },
+    "office visit": {
+        "procedure": "Office Visit - Established Patient",
+        "hcpcs_code": "99213",
+        "average_cost": 150,
+        "low_estimate": 100,
+        "high_estimate": 300,
+        "medicare_rate": 93,
+        "factors": ["Complexity level", "Time spent", "Geographic location"]
+    }
+}
+
+# Sample facility comparison data
+FACILITY_COSTS = {
+    "45378": [  # Colonoscopy
+        {"facility": "University Hospital", "gross_charge": 4500, "negotiated_rate": 2800, "cash_price": 2500},
+        {"facility": "Community Medical Center", "gross_charge": 3200, "negotiated_rate": 2100, "cash_price": 1800},
+        {"facility": "Ambulatory Surgery Center", "gross_charge": 2800, "negotiated_rate": 1600, "cash_price": 1500},
+    ],
+    "70450": [  # CT Scan
+        {"facility": "University Hospital", "gross_charge": 1800, "negotiated_rate": 820, "cash_price": 600},
+        {"facility": "Community Medical Center", "gross_charge": 1200, "negotiated_rate": 650, "cash_price": 500},
+        {"facility": "Imaging Center", "gross_charge": 900, "negotiated_rate": 450, "cash_price": 400},
+    ],
+    "70553": [  # MRI
+        {"facility": "University Hospital", "gross_charge": 3500, "negotiated_rate": 1800, "cash_price": 1500},
+        {"facility": "Community Medical Center", "gross_charge": 2800, "negotiated_rate": 1400, "cash_price": 1200},
+        {"facility": "Imaging Center", "gross_charge": 2200, "negotiated_rate": 1000, "cash_price": 900},
+    ]
+}
+
+# Procedure code database
+PROCEDURE_CODES = [
+    {"code": "99213", "description": "Office visit, established patient, level 3", "category": "Evaluation & Management"},
+    {"code": "99214", "description": "Office visit, established patient, level 4", "category": "Evaluation & Management"},
+    {"code": "99215", "description": "Office visit, established patient, level 5", "category": "Evaluation & Management"},
+    {"code": "99203", "description": "Office visit, new patient, level 3", "category": "Evaluation & Management"},
+    {"code": "45378", "description": "Colonoscopy, diagnostic", "category": "Gastroenterology"},
+    {"code": "43239", "description": "Upper GI endoscopy with biopsy", "category": "Gastroenterology"},
+    {"code": "70450", "description": "CT scan, head without contrast", "category": "Radiology"},
+    {"code": "70553", "description": "MRI brain with contrast", "category": "Radiology"},
+    {"code": "71046", "description": "Chest X-ray, 2 views", "category": "Radiology"},
+    {"code": "80053", "description": "Comprehensive metabolic panel", "category": "Laboratory"},
+    {"code": "85025", "description": "Complete blood count with differential", "category": "Laboratory"},
+    {"code": "82947", "description": "Glucose blood test", "category": "Laboratory"},
+    {"code": "83036", "description": "Hemoglobin A1C", "category": "Laboratory"},
+]
+
+
+@register_provider("cms_pricing")
+class CMSPricingProvider(BaseProvider):
+    """Provider for Medicare pricing and cost transparency data."""
+
+    def __init__(self, client: httpx.AsyncClient):
+        super().__init__("cms_pricing", client)
+
+    async def initialize(self) -> None:
+        """Initialize CMS pricing provider."""
+        logger.info("CMS Pricing provider initialized")
+
+    def get_tools(self) -> List[Callable]:
+        """Return all CMS pricing tools."""
+        return [
+            self.cms_get_medicare_fee,
+            self.cms_estimate_procedure_cost,
+            self.cms_search_procedure_codes,
+            self.cms_compare_facility_costs,
+            self.cms_estimate_out_of_pocket,
+        ]
+
+    @safe_json_return
+    async def cms_get_medicare_fee(
+        self,
+        hcpcs_code: str,
+        locality: str = "National"
+    ) -> Dict[str, Any]:
+        """
+        Get Medicare fee schedule pricing for a procedure by HCPCS/CPT code.
+
+        Args:
+            hcpcs_code: HCPCS procedure code (e.g., "99213", "43239")
+            locality: Geographic locality (default: "National")
+
+        Returns:
+            Medicare reimbursement rates with facility vs non-facility pricing
+        """
+        result = MEDICARE_FEE_SCHEDULE.get(hcpcs_code, {
+            "description": f"Procedure {hcpcs_code}",
+            "medicare_payment": 150.00,
+            "facility_rate": 150.00,
+            "non_facility_rate": 200.00,
+            "rvu": 3.0,
+            "note": "Sample data - actual rates vary by locality and year"
+        })
+
+        result["hcpcs_code"] = hcpcs_code
+        result["locality"] = locality
+        result["year"] = 2024
+
+        return result
+
+    @safe_json_return
+    async def cms_estimate_procedure_cost(
+        self,
+        procedure_name: str,
+        region: str = "National",
+        insurance_type: str = "Medicare"
+    ) -> Dict[str, Any]:
+        """
+        Estimate typical cost range for a procedure in a geographic region.
+
+        Args:
+            procedure_name: Common procedure name (e.g., "colonoscopy", "MRI", "blood test")
+            region: Geographic region (default: "National")
+            insurance_type: Type of insurance (Medicare, Commercial, Cash)
+
+        Returns:
+            Estimated cost range with average, low, and high estimates
+        """
+        procedure_key = procedure_name.lower()
+        result = PROCEDURE_COST_ESTIMATES.get(procedure_key, {
+            "procedure": procedure_name.title(),
+            "hcpcs_code": "Unknown",
+            "average_cost": 500,
+            "low_estimate": 200,
+            "high_estimate": 1000,
+            "medicare_rate": 150,
+            "factors": ["Procedure complexity", "Facility type", "Geographic location"]
+        })
+
+        result["region"] = region
+        result["insurance_type"] = insurance_type
+        result["note"] = "Estimates based on national averages. Actual costs vary significantly by provider and location."
+
+        # Adjust for insurance type
+        if insurance_type.lower() == "commercial":
+            result["estimated_patient_cost"] = result["average_cost"]
+        elif insurance_type.lower() == "medicare":
+            result["estimated_patient_cost"] = result["medicare_rate"] * 1.2  # 20% coinsurance
+        elif insurance_type.lower() == "cash":
+            result["estimated_patient_cost"] = result["average_cost"] * 0.7  # Cash discount
+
+        return result
+
+    @safe_json_return
+    async def cms_search_procedure_codes(
+        self,
+        query: str,
+        limit: int = 10
+    ) -> Dict[str, Any]:
+        """
+        Search for HCPCS/CPT codes by procedure description or category.
+
+        Args:
+            query: Search term (e.g., "office visit", "colonoscopy")
+            limit: Maximum number of results (default: 10)
+
+        Returns:
+            List of matching procedures with codes, descriptions, and categories
+        """
+        query_lower = query.lower()
+        matches = [
+            p for p in PROCEDURE_CODES
+            if query_lower in p["description"].lower() or query_lower in p["category"].lower()
+        ]
+
+        return {
+            "query": query,
+            "results": matches[:limit],
+            "total": len(matches)
+        }
+
+    @safe_json_return
+    async def cms_compare_facility_costs(
+        self,
+        procedure_code: str,
+        facilities: List[str] = None
+    ) -> Dict[str, Any]:
+        """
+        Compare procedure costs across different healthcare facilities.
+
+        Args:
+            procedure_code: HCPCS/CPT code
+            facilities: List of facility names (uses sample data if not provided)
+
+        Returns:
+            Cost comparison showing gross charges, negotiated rates, and cash prices
+        """
+        facilities_data = FACILITY_COSTS.get(procedure_code, [
+            {"facility": "Sample Facility A", "gross_charge": 2000, "negotiated_rate": 1200, "cash_price": 1000},
+            {"facility": "Sample Facility B", "gross_charge": 1500, "negotiated_rate": 900, "cash_price": 800},
+            {"facility": "Sample Facility C", "gross_charge": 1800, "negotiated_rate": 1000, "cash_price": 900},
+        ])
+
+        # Calculate savings
+        cash_prices = [f["cash_price"] for f in facilities_data]
+        lowest_cash = min(cash_prices)
+        highest_cash = max(cash_prices)
+        potential_savings = highest_cash - lowest_cash
+
+        return {
+            "procedure_code": procedure_code,
+            "facilities": facilities_data,
+            "comparison": {
+                "lowest_cash_price": lowest_cash,
+                "highest_cash_price": highest_cash,
+                "potential_savings": potential_savings,
+                "average_cash_price": sum(cash_prices) / len(cash_prices)
+            },
+            "note": "Sample data for demonstration. Actual facility prices vary. Patients should request price estimates directly."
+        }
+
+    @safe_json_return
+    async def cms_estimate_out_of_pocket(
+        self,
+        procedure_code: str,
+        insurance_scenario: str = "Medicare",
+        deductible_met: bool = False
+    ) -> Dict[str, Any]:
+        """
+        Estimate patient out-of-pocket costs for common insurance scenarios.
+
+        Args:
+            procedure_code: HCPCS/CPT code
+            insurance_scenario: Insurance type (Medicare, Commercial High Deductible, Commercial PPO, Cash)
+            deductible_met: Whether annual deductible has been met
+
+        Returns:
+            Estimated patient financial responsibility
+        """
+        # Get base Medicare cost
+        medicare_data = await self.cms_get_medicare_fee(procedure_code)
+        base_cost = medicare_data.get("medicare_payment", 200)
+
+        # Define insurance scenarios
+        scenarios = {
+            "Medicare": {
+                "coinsurance": 0.20,
+                "copay": 0,
+                "description": "Medicare Part B (20% coinsurance after deductible)",
+                "deductible": 240  # 2024 Part B deductible
+            },
+            "Commercial PPO": {
+                "coinsurance": 0.20,
+                "copay": 0,
+                "description": "Commercial PPO (20% coinsurance after deductible)",
+                "deductible": 2000,
+                "multiplier": 2.5  # Commercial pays ~2.5x Medicare
+            },
+            "Commercial High Deductible": {
+                "coinsurance": 0.30,
+                "copay": 0,
+                "description": "High Deductible Health Plan (30% coinsurance)",
+                "deductible": 5000,
+                "multiplier": 2.5
+            },
+            "Cash": {
+                "coinsurance": 0,
+                "copay": 0,
+                "description": "Cash/Self-pay (negotiated rate)",
+                "deductible": 0,
+                "multiplier": 1.8  # Cash discount from commercial
+            }
+        }
+
+        scenario = scenarios.get(insurance_scenario, scenarios["Medicare"])
+
+        # Calculate costs
+        allowed_amount = base_cost * scenario.get("multiplier", 1.0)
+
+        if deductible_met:
+            patient_responsibility = allowed_amount * scenario["coinsurance"] + scenario["copay"]
+        else:
+            # Patient pays deductible first, then coinsurance on remainder
+            if allowed_amount <= scenario["deductible"]:
+                patient_responsibility = allowed_amount
+            else:
+                remaining = allowed_amount - scenario["deductible"]
+                patient_responsibility = scenario["deductible"] + (remaining * scenario["coinsurance"])
+
+        return {
+            "procedure_code": procedure_code,
+            "procedure_description": medicare_data.get("description", "Unknown procedure"),
+            "insurance_scenario": insurance_scenario,
+            "deductible_met": deductible_met,
+            "cost_breakdown": {
+                "allowed_amount": round(allowed_amount, 2),
+                "deductible": scenario["deductible"],
+                "coinsurance_rate": f"{int(scenario['coinsurance'] * 100)}%",
+                "estimated_patient_responsibility": round(patient_responsibility, 2),
+                "insurance_pays": round(allowed_amount - patient_responsibility, 2)
+            },
+            "scenario_details": scenario["description"],
+            "note": "Estimates only. Actual costs depend on specific plan benefits and negotiated rates."
+        }

providers/openfda_provider.py

@@ -0,0 +1,573 @@
+"""OpenFDA API Provider for adverse events, drug labels, recalls, and more."""
+
+import re
+import logging
+from typing import List, Dict, Any, Callable, Optional
+import httpx
+from functools import lru_cache
+
+from ..core.base_provider import BaseProvider
+from ..core.decorators import safe_json_return, with_retry
+from . import register_provider
+
+logger = logging.getLogger(__name__)
+
+# OpenFDA API endpoints
+OPENFDA_DRUG_EVENT = "https://api.fda.gov/drug/event.json"
+OPENFDA_DRUG_LABEL = "https://api.fda.gov/drug/label.json"
+OPENFDA_DRUG_NDC = "https://api.fda.gov/drug/ndc.json"
+OPENFDA_DRUG_ENFORCEMENT = "https://api.fda.gov/drug/enforcement.json"
+OPENFDA_DRUG_DRUGSFDA = "https://api.fda.gov/drug/drugsfda.json"
+OPENFDA_DEVICE_EVENT = "https://api.fda.gov/device/event.json"
+OPENFDA_DEVICE_ENFORCEMENT = "https://api.fda.gov/device/enforcement.json"
+OPENFDA_DEVICE_CLASSIFICATION = "https://api.fda.gov/device/classification.json"
+OPENFDA_DEVICE_510K = "https://api.fda.gov/device/510k.json"
+OPENFDA_DEVICE_PMA = "https://api.fda.gov/device/pma.json"
+OPENFDA_FOOD_ENFORCEMENT = "https://api.fda.gov/food/enforcement.json"
+OPENFDA_FOOD_EVENT = "https://api.fda.gov/food/event.json"
+
+
+def normalize_drug_name(raw_name: str) -> str:
+    """Extract base drug name from full medication name."""
+    if not raw_name:
+        return ""
+
+    cleaned = raw_name
+    patterns = [
+        r'\s+\d+(?:\.\d+)?\s*(MG|MCG|ML|G|%)',
+        r'\s+Oral\s+',
+        r'\s+Tablet\s*',
+        r'\s+Capsule\s*',
+        r'\s+Injectable\s*',
+        r'\s+Solution\s*',
+        r'\s+Suspension\s*'
+    ]
+
+    for pattern in patterns:
+        cleaned = re.split(pattern, cleaned, flags=re.IGNORECASE)[0]
+
+    base_name = cleaned.strip().split()[0] if cleaned.strip() else cleaned.strip()
+    return base_name
+
+
+@register_provider("openfda")
+class OpenFDAProvider(BaseProvider):
+    """Provider for OpenFDA APIs."""
+
+    def __init__(self, client: httpx.AsyncClient, api_key: Optional[str] = None):
+        super().__init__("openfda", client)
+        self.api_key = api_key
+
+    async def initialize(self) -> None:
+        """Initialize OpenFDA provider."""
+        logger.info("OpenFDA provider initialized")
+
+    def get_tools(self) -> List[Callable]:
+        """Return all OpenFDA tools."""
+        return [
+            self.openfda_get_adverse_event_summary,
+            self.openfda_fetch_adverse_events,
+            self.openfda_top_reactions,
+            self.openfda_search_drug_labels,
+            self.openfda_search_ndc,
+            self.openfda_search_drug_recalls,
+            self.openfda_search_drugs_fda,
+            self.openfda_search_device_events,
+            self.openfda_search_device_recalls,
+            self.openfda_search_device_classifications,
+            self.openfda_search_510k,
+            self.openfda_search_pma,
+            self.openfda_search_food_recalls,
+            self.openfda_search_food_events,
+        ]
+
+    def _build_params(self, search: str, limit: int = 10) -> Dict[str, Any]:
+        """Build query parameters with optional API key."""
+        params = {"search": search, "limit": min(limit, 100)}
+        if self.api_key:
+            params["api_key"] = self.api_key
+        return params
+
+    @with_retry
+    async def _fetch_fda_data(self, url: str, params: Dict[str, Any]) -> Dict[str, Any]:
+        """Fetch data from OpenFDA API with retry logic."""
+        response = await self.client.get(url, params=params)
+        response.raise_for_status()
+        return response.json()
+
+    # Drug Adverse Events
+    @safe_json_return
+    async def openfda_get_adverse_event_summary(self, drug_name: str) -> Dict[str, Any]:
+        """
+        Get high-level adverse event summary for a medication from FAERS database.
+
+        Args:
+            drug_name: Name of the medication (generic or brand name)
+
+        Returns:
+            Summary with total reports, serious reports, and top reactions
+        """
+        clean_name = normalize_drug_name(drug_name)
+        logger.info(f"FDA adverse event query: '{drug_name}' -> '{clean_name}'")
+
+        # Query for count and reactions
+        search_query = f'patient.drug.medicinalproduct:"{clean_name}"'
+        params = self._build_params(search_query, limit=1)
+        params["count"] = "patient.reaction.reactionmeddrapt.exact"
+
+        data = await self._fetch_fda_data(OPENFDA_DRUG_EVENT, params)
+
+        total_reports = data.get("meta", {}).get("results", {}).get("total", 0)
+        reactions = data.get("results", [])[:5]
+
+        top_reactions = [
+            {"reaction": r.get("term", "Unknown"), "count": r.get("count", 0)}
+            for r in reactions
+        ]
+
+        # Get serious event count
+        serious_query = f'{search_query}+AND+serious:1'
+        serious_params = self._build_params(serious_query, limit=1)
+        try:
+            serious_data = await self._fetch_fda_data(OPENFDA_DRUG_EVENT, serious_params)
+            serious_reports = serious_data.get("meta", {}).get("results", {}).get("total", 0)
+        except Exception:
+            serious_reports = 0
+
+        return {
+            "drug": clean_name,
+            "total_reports": total_reports,
+            "serious_reports": serious_reports,
+            "top_reactions": top_reactions
+        }
+
+    @safe_json_return
+    async def openfda_fetch_adverse_events(
+        self,
+        drug_name: str,
+        limit: int = 25,
+        max_pages: int = 1
+    ) -> Dict[str, Any]:
+        """
+        Fetch raw adverse event reports from OpenFDA with pagination.
+
+        Args:
+            drug_name: Name of the medication
+            limit: Number of events per page (max 100)
+            max_pages: Number of pages to fetch
+
+        Returns:
+            List of adverse event reports with metadata
+        """
+        clean_name = normalize_drug_name(drug_name)
+        search_query = f'patient.drug.medicinalproduct:"{clean_name}"'
+
+        all_events = []
+        for page in range(max_pages):
+            params = self._build_params(search_query, limit=min(limit, 100))
+            params["skip"] = page * limit
+
+            try:
+                data = await self._fetch_fda_data(OPENFDA_DRUG_EVENT, params)
+                results = data.get("results", [])
+
+                for result in results:
+                    reactions = result.get("patient", {}).get("reaction", [])
+                    reaction_list = [r.get("reactionmeddrapt", "Unknown") for r in reactions]
+
+                    all_events.append({
+                        "safety_report_id": result.get("safetyreportid", "Unknown"),
+                        "receive_date": result.get("receivedate", "Unknown"),
+                        "serious": result.get("serious", 0) == 1,
+                        "reactions": reaction_list[:5],
+                        "patient_age": result.get("patient", {}).get("patientonsetage", "Unknown"),
+                        "patient_sex": result.get("patient", {}).get("patientsex", "Unknown")
+                    })
+
+                if len(results) < limit:
+                    break
+            except Exception as e:
+                logger.warning(f"Error fetching page {page}: {e}")
+                break
+
+        return {
+            "drug": clean_name,
+            "events": all_events,
+            "total_fetched": len(all_events)
+        }
+
+    @safe_json_return
+    async def openfda_top_reactions(self, drug_name: str) -> Dict[str, Any]:
+        """
+        Get top 5 most commonly reported adverse reactions for a medication.
+
+        Args:
+            drug_name: Name of the medication
+
+        Returns:
+            Top 5 reactions with counts
+        """
+        clean_name = normalize_drug_name(drug_name)
+        search_query = f'patient.drug.medicinalproduct:"{clean_name}"'
+
+        params = self._build_params(search_query, limit=1)
+        params["count"] = "patient.reaction.reactionmeddrapt.exact"
+
+        data = await self._fetch_fda_data(OPENFDA_DRUG_EVENT, params)
+        reactions = data.get("results", [])[:5]
+
+        return {
+            "drug": clean_name,
+            "top_reactions": [
+                {"reaction": r.get("term", "Unknown"), "count": r.get("count", 0)}
+                for r in reactions
+            ]
+        }
+
+    # Drug Labels
+    @safe_json_return
+    async def openfda_search_drug_labels(self, query: str, limit: int = 10) -> Dict[str, Any]:
+        """
+        Search FDA drug labeling information (warnings, indications, dosage).
+
+        Args:
+            query: Drug name, active ingredient, or condition
+            limit: Maximum results (max 100)
+
+        Returns:
+            Drug label information
+        """
+        params = self._build_params(query, limit)
+        data = await self._fetch_fda_data(OPENFDA_DRUG_LABEL, params)
+
+        results = data.get("results", [])
+        labels = []
+
+        for result in results:
+            openfda = result.get("openfda", {})
+            labels.append({
+                "brand_name": openfda.get("brand_name", ["Unknown"])[0],
+                "generic_name": openfda.get("generic_name", ["Unknown"])[0],
+                "manufacturer": openfda.get("manufacturer_name", ["Unknown"])[0],
+                "purpose": result.get("purpose", ["Not specified"])[0] if result.get("purpose") else "Not specified",
+                "warnings": result.get("warnings", ["Not specified"])[0][:500] if result.get("warnings") else "Not specified",
+                "indications_and_usage": result.get("indications_and_usage", ["Not specified"])[0][:500] if result.get("indications_and_usage") else "Not specified"
+            })
+
+        return {"results": labels, "total": len(labels)}
+
+    # NDC Directory
+    @safe_json_return
+    async def openfda_search_ndc(self, query: str, limit: int = 10) -> Dict[str, Any]:
+        """
+        Search National Drug Code (NDC) directory for drug product information.
+
+        Args:
+            query: Brand name, generic name, or NDC number
+            limit: Maximum results (max 100)
+
+        Returns:
+            NDC product information
+        """
+        params = self._build_params(query, limit)
+        data = await self._fetch_fda_data(OPENFDA_DRUG_NDC, params)
+
+        results = data.get("results", [])
+        products = []
+
+        for result in results:
+            products.append({
+                "product_ndc": result.get("product_ndc", "Unknown"),
+                "brand_name": result.get("brand_name", "Unknown"),
+                "generic_name": result.get("generic_name", "Unknown"),
+                "manufacturer": result.get("labeler_name", "Unknown"),
+                "dosage_form": result.get("dosage_form", "Unknown"),
+                "route": result.get("route", ["Unknown"])[0] if result.get("route") else "Unknown",
+                "marketing_status": result.get("marketing_status", "Unknown")
+            })
+
+        return {"results": products, "total": len(products)}
+
+    # Drug Recalls
+    @safe_json_return
+    async def openfda_search_drug_recalls(self, query: str, limit: int = 10) -> Dict[str, Any]:
+        """
+        Search FDA drug recall and enforcement reports.
+
+        Args:
+            query: Drug name, manufacturer, or reason for recall
+            limit: Maximum results (max 100)
+
+        Returns:
+            Drug recall information
+        """
+        params = self._build_params(query, limit)
+        data = await self._fetch_fda_data(OPENFDA_DRUG_ENFORCEMENT, params)
+
+        results = data.get("results", [])
+        recalls = []
+
+        for result in results:
+            recalls.append({
+                "product_description": result.get("product_description", "Unknown"),
+                "reason_for_recall": result.get("reason_for_recall", "Unknown"),
+                "classification": result.get("classification", "Unknown"),
+                "status": result.get("status", "Unknown"),
+                "recall_date": result.get("recall_initiation_date", "Unknown"),
+                "recalling_firm": result.get("recalling_firm", "Unknown")
+            })
+
+        return {"results": recalls, "total": len(recalls)}
+
+    # Drugs@FDA
+    @safe_json_return
+    async def openfda_search_drugs_fda(self, query: str, limit: int = 10) -> Dict[str, Any]:
+        """
+        Search Drugs@FDA database for approved drug products and applications.
+
+        Args:
+            query: Drug name or active ingredient
+            limit: Maximum results (max 100)
+
+        Returns:
+            Approved drug information
+        """
+        params = self._build_params(query, limit)
+        data = await self._fetch_fda_data(OPENFDA_DRUG_DRUGSFDA, params)
+
+        results = data.get("results", [])
+        drugs = []
+
+        for result in results:
+            products = result.get("products", [])
+            for product in products:
+                drugs.append({
+                    "application_number": result.get("application_number", "Unknown"),
+                    "sponsor_name": result.get("sponsor_name", "Unknown"),
+                    "brand_name": product.get("brand_name", "Unknown"),
+                    "active_ingredients": product.get("active_ingredients", []),
+                    "dosage_form": product.get("dosage_form", "Unknown"),
+                    "route": product.get("route", "Unknown"),
+                    "marketing_status": product.get("marketing_status", "Unknown")
+                })
+
+        return {"results": drugs, "total": len(drugs)}
+
+    # Device Events
+    @safe_json_return
+    async def openfda_search_device_events(self, query: str, limit: int = 10) -> Dict[str, Any]:
+        """
+        Search medical device adverse event reports.
+
+        Args:
+            query: Device name or brand
+            limit: Maximum results (max 100)
+
+        Returns:
+            Device adverse event information
+        """
+        params = self._build_params(query, limit)
+        data = await self._fetch_fda_data(OPENFDA_DEVICE_EVENT, params)
+
+        results = data.get("results", [])
+        events = []
+
+        for result in results:
+            device = result.get("device", [{}])[0]
+            events.append({
+                "report_number": result.get("report_number", "Unknown"),
+                "date_received": result.get("date_received", "Unknown"),
+                "device_name": device.get("brand_name", "Unknown"),
+                "manufacturer": device.get("manufacturer_d_name", "Unknown"),
+                "event_type": result.get("event_type", "Unknown"),
+                "device_problem": device.get("device_problem_codes", ["Unknown"])[0] if device.get("device_problem_codes") else "Unknown"
+            })
+
+        return {"results": events, "total": len(events)}
+
+    # Device Recalls
+    @safe_json_return
+    async def openfda_search_device_recalls(self, query: str, limit: int = 10) -> Dict[str, Any]:
+        """
+        Search medical device recall and enforcement reports.
+
+        Args:
+            query: Device name, manufacturer, or reason
+            limit: Maximum results (max 100)
+
+        Returns:
+            Device recall information
+        """
+        params = self._build_params(query, limit)
+        data = await self._fetch_fda_data(OPENFDA_DEVICE_ENFORCEMENT, params)
+
+        results = data.get("results", [])
+        recalls = []
+
+        for result in results:
+            recalls.append({
+                "product_description": result.get("product_description", "Unknown"),
+                "reason_for_recall": result.get("reason_for_recall", "Unknown"),
+                "classification": result.get("classification", "Unknown"),
+                "status": result.get("status", "Unknown"),
+                "recall_date": result.get("recall_initiation_date", "Unknown"),
+                "recalling_firm": result.get("recalling_firm", "Unknown")
+            })
+
+        return {"results": recalls, "total": len(recalls)}
+
+    # Device Classifications
+    @safe_json_return
+    async def openfda_search_device_classifications(self, query: str, limit: int = 10) -> Dict[str, Any]:
+        """
+        Search medical device classification database.
+
+        Args:
+            query: Device type or classification
+            limit: Maximum results (max 100)
+
+        Returns:
+            Device classification information
+        """
+        params = self._build_params(query, limit)
+        data = await self._fetch_fda_data(OPENFDA_DEVICE_CLASSIFICATION, params)
+
+        results = data.get("results", [])
+        classifications = []
+
+        for result in results:
+            classifications.append({
+                "device_name": result.get("device_name", "Unknown"),
+                "device_class": result.get("device_class", "Unknown"),
+                "medical_specialty": result.get("medical_specialty_description", "Unknown"),
+                "regulation_number": result.get("regulation_number", "Unknown"),
+                "product_code": result.get("product_code", "Unknown")
+            })
+
+        return {"results": classifications, "total": len(classifications)}
+
+    # 510(k) Clearances
+    @safe_json_return
+    async def openfda_search_510k(self, query: str, limit: int = 10) -> Dict[str, Any]:
+        """
+        Search FDA 510(k) premarket clearance database.
+
+        Args:
+            query: Device name or manufacturer
+            limit: Maximum results (max 100)
+
+        Returns:
+            510(k) clearance information
+        """
+        params = self._build_params(query, limit)
+        data = await self._fetch_fda_data(OPENFDA_DEVICE_510K, params)
+
+        results = data.get("results", [])
+        clearances = []
+
+        for result in results:
+            clearances.append({
+                "k_number": result.get("k_number", "Unknown"),
+                "device_name": result.get("device_name", "Unknown"),
+                "applicant": result.get("applicant", "Unknown"),
+                "clearance_date": result.get("date_received", "Unknown"),
+                "decision_description": result.get("decision_description", "Unknown"),
+                "product_code": result.get("product_code", "Unknown")
+            })
+
+        return {"results": clearances, "total": len(clearances)}
+
+    # PMA Approvals
+    @safe_json_return
+    async def openfda_search_pma(self, query: str, limit: int = 10) -> Dict[str, Any]:
+        """
+        Search FDA premarket approval (PMA) database.
+
+        Args:
+            query: Device name or manufacturer
+            limit: Maximum results (max 100)
+
+        Returns:
+            PMA approval information
+        """
+        params = self._build_params(query, limit)
+        data = await self._fetch_fda_data(OPENFDA_DEVICE_PMA, params)
+
+        results = data.get("results", [])
+        approvals = []
+
+        for result in results:
+            approvals.append({
+                "pma_number": result.get("pma_number", "Unknown"),
+                "device_name": result.get("device_name", "Unknown"),
+                "applicant": result.get("applicant", "Unknown"),
+                "approval_date": result.get("date_received", "Unknown"),
+                "decision_description": result.get("decision_description", "Unknown"),
+                "product_code": result.get("product_code", "Unknown")
+            })
+
+        return {"results": approvals, "total": len(approvals)}
+
+    # Food Recalls
+    @safe_json_return
+    async def openfda_search_food_recalls(self, query: str, limit: int = 10) -> Dict[str, Any]:
+        """
+        Search FDA food recall and enforcement reports.
+
+        Args:
+            query: Food product or reason for recall
+            limit: Maximum results (max 100)
+
+        Returns:
+            Food recall information
+        """
+        params = self._build_params(query, limit)
+        data = await self._fetch_fda_data(OPENFDA_FOOD_ENFORCEMENT, params)
+
+        results = data.get("results", [])
+        recalls = []
+
+        for result in results:
+            recalls.append({
+                "product_description": result.get("product_description", "Unknown"),
+                "reason_for_recall": result.get("reason_for_recall", "Unknown"),
+                "classification": result.get("classification", "Unknown"),
+                "status": result.get("status", "Unknown"),
+                "recall_date": result.get("recall_initiation_date", "Unknown"),
+                "recalling_firm": result.get("recalling_firm", "Unknown")
+            })
+
+        return {"results": recalls, "total": len(recalls)}
+
+    # Food Events
+    @safe_json_return
+    async def openfda_search_food_events(self, query: str, limit: int = 10) -> Dict[str, Any]:
+        """
+        Search FDA food adverse event reports.
+
+        Args:
+            query: Food product or reaction
+            limit: Maximum results (max 100)
+
+        Returns:
+            Food adverse event information
+        """
+        params = self._build_params(query, limit)
+        data = await self._fetch_fda_data(OPENFDA_FOOD_EVENT, params)
+
+        results = data.get("results", [])
+        events = []
+
+        for result in results:
+            products = result.get("products", [{}])
+            reactions = result.get("reactions", [])
+
+            events.append({
+                "report_number": result.get("report_number", "Unknown"),
+                "date_started": result.get("date_started", "Unknown"),
+                "products": [p.get("name_brand", "Unknown") for p in products],
+                "reactions": [r.get("reaction", "Unknown") for r in reactions][:5],
+                "outcomes": result.get("outcomes", ["Unknown"])
+            })
+
+        return {"results": events, "total": len(events)}

requirements.txt

@@ -0,0 +1,6 @@
+mcp>=1.0.0
+httpx>=0.28.0
+pydantic>=2.11.0
+tenacity>=9.0.0
+fastapi>=0.104.0
+uvicorn[standard]>=0.24.0

server.py

@@ -0,0 +1,223 @@
+"""Healthcare API MCP Server - Public APIs for FDA, Clinical Guidelines, and CMS."""
+
+import asyncio
+import logging
+from typing import List, Dict, Any
+
+from mcp.server import Server
+from mcp.server.stdio import stdio_server
+from mcp.types import Tool, TextContent
+
+from .config import MCPConfig
+from .core import create_http_client
+from .providers import get_provider_class
+from .providers.openfda_provider import OpenFDAProvider
+from .providers.clinical_guidelines_provider import ClinicalGuidelinesProvider
+from .providers.cms_pricing_provider import CMSPricingProvider
+
+# Setup logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+# Initialize MCP server
+app = Server("healthcare-api-mcp")
+
+# Global state
+providers = {}
+http_client = None
+
+
+def create_tool_schema(func: callable, provider_name: str) -> Tool:
+    """Create MCP tool schema from provider function."""
+    # Get function metadata
+    doc = func.__doc__ or ""
+    lines = [line.strip() for line in doc.split('\n') if line.strip()]
+
+    # Extract description (first non-empty line)
+    description = lines[0] if lines else func.__name__
+
+    # Extract parameter info from docstring
+    # Parse Args section
+    params_start = None
+    params_end = None
+    for i, line in enumerate(lines):
+        if line.startswith('Args:'):
+            params_start = i + 1
+        elif params_start and line.startswith('Returns:'):
+            params_end = i
+            break
+
+    # Build input schema
+    properties = {}
+    required = []
+
+    if params_start and params_end:
+        for line in lines[params_start:params_end]:
+            if ':' in line:
+                param_info = line.split(':', 1)
+                param_name = param_info[0].strip()
+                param_desc = param_info[1].strip() if len(param_info) > 1 else ""
+
+                # Determine type from function annotations
+                func_params = func.__annotations__
+                param_type = "string"  # default
+
+                if param_name in func_params:
+                    annotation = func_params[param_name]
+                    if annotation == int:
+                        param_type = "integer"
+                    elif annotation == bool:
+                        param_type = "boolean"
+                    elif annotation == float:
+                        param_type = "number"
+                    elif hasattr(annotation, '__origin__'):
+                        # Handle List[str], etc.
+                        if annotation.__origin__ == list:
+                            param_type = "array"
+
+                properties[param_name] = {
+                    "type": param_type,
+                    "description": param_desc
+                }
+
+                # Check if parameter has no default value (is required)
+                import inspect
+                sig = inspect.signature(func)
+                if param_name in sig.parameters:
+                    param = sig.parameters[param_name]
+                    if param.default == inspect.Parameter.empty and param_name != 'self':
+                        required.append(param_name)
+
+    # Create tool schema
+    tool = Tool(
+        name=func.__name__,
+        description=description,
+        inputSchema={
+            "type": "object",
+            "properties": properties,
+            "required": required
+        }
+    )
+
+    return tool
+
+
+@app.list_tools()
+async def list_tools() -> List[Tool]:
+    """List all available tools from enabled providers."""
+    tools = []
+
+    for provider_name, provider in providers.items():
+        provider_tools = provider.get_tools()
+        for tool_func in provider_tools:
+            tool_schema = create_tool_schema(tool_func, provider_name)
+            tools.append(tool_schema)
+
+    logger.info(f"Listed {len(tools)} tools from {len(providers)} providers")
+    return tools
+
+
+@app.call_tool()
+async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
+    """Execute a tool from any enabled provider."""
+    logger.info(f"Calling tool: {name} with arguments: {arguments}")
+
+    # Find the tool in providers
+    for provider_name, provider in providers.items():
+        provider_tools = provider.get_tools()
+        for tool_func in provider_tools:
+            if tool_func.__name__ == name:
+                try:
+                    # Call the tool function
+                    result = await tool_func(**arguments)
+
+                    # Return as TextContent
+                    return [TextContent(
+                        type="text",
+                        text=str(result)
+                    )]
+                except Exception as e:
+                    logger.exception(f"Error calling tool {name}")
+                    error_result = f'{{"error": "{str(e)}", "error_type": "{type(e).__name__}"}}'
+                    return [TextContent(
+                        type="text",
+                        text=error_result
+                    )]
+
+    # Tool not found
+    error_msg = f'{{"error": "Tool not found: {name}"}}'
+    return [TextContent(type="text", text=error_msg)]
+
+
+async def main():
+    """Main entry point for the MCP server."""
+    global providers, http_client
+
+    # Load configuration
+    config = MCPConfig.from_env()
+
+    # Set logging level
+    logging.getLogger().setLevel(config.log_level)
+
+    logger.info("Starting Healthcare API MCP Server")
+    logger.info(f"Enabled providers: {config.enabled_providers}")
+
+    # Create shared HTTP client
+    http_client = create_http_client()
+
+    # Initialize providers
+    provider_classes = {
+        "openfda": OpenFDAProvider,
+        "clinical_guidelines": ClinicalGuidelinesProvider,
+        "cms_pricing": CMSPricingProvider
+    }
+
+    for provider_name in config.enabled_providers:
+        if provider_name not in provider_classes:
+            logger.warning(f"Unknown provider: {provider_name}")
+            continue
+
+        try:
+            provider_class = provider_classes[provider_name]
+
+            # Initialize provider with appropriate parameters
+            if provider_name == "openfda":
+                provider = provider_class(http_client, api_key=config.openfda_api_key)
+            else:
+                provider = provider_class(http_client)
+
+            await provider.initialize()
+            providers[provider_name] = provider
+            logger.info(f"Initialized provider: {provider_name}")
+        except Exception as e:
+            logger.error(f"Failed to initialize provider {provider_name}: {e}")
+
+    if not providers:
+        logger.error("No providers initialized. Exiting.")
+        return
+
+    logger.info(f"Successfully initialized {len(providers)} provider(s)")
+
+    # Run the server with stdio transport
+    try:
+        async with stdio_server() as (read_stream, write_stream):
+            logger.info("MCP server running on stdio")
+            await app.run(
+                read_stream,
+                write_stream,
+                app.create_initialization_options()
+            )
+    finally:
+        # Cleanup
+        logger.info("Shutting down...")
+        for provider in providers.values():
+            await provider.cleanup()
+        if http_client:
+            await http_client.aclose()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

server_http.py

@@ -0,0 +1,265 @@
+"""HTTP Streaming MCP Server for Healthcare API MCP - HuggingFace Spaces deployment.
+
+Uses HTTP streaming with chunked transfer encoding instead of SSE.
+"""
+
+import asyncio
+import json
+import logging
+from typing import AsyncGenerator
+
+from fastapi import FastAPI, Request
+from fastapi.responses import StreamingResponse
+from starlette.middleware.cors import CORSMiddleware
+
+from config import MCPConfig
+from core import create_http_client
+from providers.openfda_provider import OpenFDAProvider
+from providers.clinical_guidelines_provider import ClinicalGuidelinesProvider
+from providers.cms_pricing_provider import CMSPricingProvider
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Initialize FastAPI app
+app = FastAPI(
+    title="Healthcare API MCP",
+    description="Model Context Protocol server for public healthcare data APIs",
+    version="1.0.0"
+)
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Global state
+providers = {}
+http_client = None
+
+
+@app.on_event("startup")
+async def startup():
+    """Initialize providers on startup."""
+    global http_client, providers
+
+    config = MCPConfig.from_env()
+    logger.info(f"Starting Healthcare API MCP with providers: {config.enabled_providers}")
+
+    http_client = create_http_client()
+
+    if "openfda" in config.enabled_providers:
+        providers["openfda"] = OpenFDAProvider(http_client, config.openfda_api_key)
+        await providers["openfda"].initialize()
+        logger.info("✅ OpenFDA provider initialized")
+
+    if "clinical_guidelines" in config.enabled_providers:
+        providers["clinical_guidelines"] = ClinicalGuidelinesProvider(http_client)
+        await providers["clinical_guidelines"].initialize()
+        logger.info("✅ Clinical Guidelines provider initialized")
+
+    if "cms_pricing" in config.enabled_providers:
+        providers["cms_pricing"] = CMSPricingProvider(http_client)
+        await providers["cms_pricing"].initialize()
+        logger.info("✅ CMS Pricing provider initialized")
+
+    logger.info(f"Healthcare API MCP ready with {len(providers)} provider(s)")
+
+
+@app.on_event("shutdown")
+async def shutdown():
+    """Cleanup on shutdown."""
+    global http_client, providers
+
+    logger.info("Shutting down Healthcare API MCP...")
+
+    for provider in providers.values():
+        await provider.cleanup()
+
+    if http_client:
+        await http_client.aclose()
+
+
+@app.get("/")
+async def root():
+    """Root endpoint with server info."""
+    return {
+        "name": "Healthcare API MCP",
+        "version": "1.0.0",
+        "protocol": "MCP over HTTP Streaming",
+        "providers": list(providers.keys()),
+        "endpoints": {
+            "tools": "/mcp/tools",
+            "call": "/mcp/call",
+            "stream": "/mcp/stream"
+        }
+    }
+
+
+@app.get("/health")
+async def health_check():
+    """Health check endpoint."""
+    return {
+        "status": "healthy",
+        "providers": {name: "active" for name in providers.keys()}
+    }
+
+
+@app.get("/mcp/tools")
+async def list_tools():
+    """List all available MCP tools."""
+    tools = []
+
+    for provider_name, provider in providers.items():
+        provider_tools = provider.get_tools()
+        for tool_func in provider_tools:
+            # Extract tool metadata
+            doc = tool_func.__doc__ or ""
+            description = doc.split("Args:")[0].strip() if "Args:" in doc else doc.strip()
+
+            # Parse parameters from docstring
+            params = {}
+            if "Args:" in doc:
+                args_section = doc.split("Args:")[1].split("Returns:")[0]
+                for line in args_section.split("\n"):
+                    line = line.strip()
+                    if ":" in line:
+                        param_name = line.split(":")[0].strip()
+                        param_desc = line.split(":", 1)[1].strip()
+                        params[param_name] = param_desc
+
+            tools.append({
+                "name": tool_func.__name__,
+                "provider": provider_name,
+                "description": description,
+                "parameters": params
+            })
+
+    return {"tools": tools, "total": len(tools)}
+
+
+@app.post("/mcp/call")
+async def call_tool(request: Request):
+    """Call an MCP tool and return result."""
+    body = await request.json()
+    tool_name = body.get("tool")
+    arguments = body.get("arguments", {})
+
+    if not tool_name:
+        return {"error": "Missing 'tool' parameter"}
+
+    logger.info(f"Calling tool: {tool_name} with args: {arguments}")
+
+    # Find and execute tool
+    for provider_name, provider in providers.items():
+        provider_tools = provider.get_tools()
+        for tool_func in provider_tools:
+            if tool_func.__name__ == tool_name:
+                try:
+                    result = await tool_func(**arguments)
+                    return {
+                        "success": True,
+                        "tool": tool_name,
+                        "result": result
+                    }
+                except Exception as e:
+                    logger.exception(f"Error calling tool {tool_name}")
+                    return {
+                        "success": False,
+                        "tool": tool_name,
+                        "error": str(e),
+                        "error_type": type(e).__name__
+                    }
+
+    return {
+        "success": False,
+        "tool": tool_name,
+        "error": f"Tool not found: {tool_name}"
+    }
+
+
+async def stream_tool_result(tool_name: str, arguments: dict) -> AsyncGenerator[str, None]:
+    """Stream tool execution results as JSON chunks."""
+    # Send initial acknowledgment
+    yield json.dumps({
+        "type": "start",
+        "tool": tool_name,
+        "arguments": arguments
+    }) + "\n"
+
+    # Find and execute tool
+    tool_found = False
+    for provider_name, provider in providers.items():
+        provider_tools = provider.get_tools()
+        for tool_func in provider_tools:
+            if tool_func.__name__ == tool_name:
+                tool_found = True
+                try:
+                    # Execute tool
+                    result = await tool_func(**arguments)
+
+                    # Stream result
+                    yield json.dumps({
+                        "type": "result",
+                        "success": True,
+                        "data": result
+                    }) + "\n"
+
+                except Exception as e:
+                    logger.exception(f"Error executing tool {tool_name}")
+                    yield json.dumps({
+                        "type": "error",
+                        "success": False,
+                        "error": str(e),
+                        "error_type": type(e).__name__
+                    }) + "\n"
+
+                break
+        if tool_found:
+            break
+
+    if not tool_found:
+        yield json.dumps({
+            "type": "error",
+            "success": False,
+            "error": f"Tool not found: {tool_name}"
+        }) + "\n"
+
+    # Send completion
+    yield json.dumps({"type": "complete"}) + "\n"
+
+
+@app.post("/mcp/stream")
+async def stream_tool_call(request: Request):
+    """Stream tool execution results using HTTP chunked transfer encoding."""
+    body = await request.json()
+    tool_name = body.get("tool")
+    arguments = body.get("arguments", {})
+
+    if not tool_name:
+        return {"error": "Missing 'tool' parameter"}
+
+    logger.info(f"Streaming tool: {tool_name} with args: {arguments}")
+
+    return StreamingResponse(
+        stream_tool_result(tool_name, arguments),
+        media_type="application/x-ndjson",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",
+        }
+    )
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(
+        app,
+        host="0.0.0.0",
+        port=7860,
+        log_level="info"
+    )