Merge pull request #1 from humandotlearning/npi-mcp-server-implementation

explanation.md

@@ -1,23 +1,46 @@
-# NPI MCP Server for CredentialWatch
+# NPI MCP Server Explanation
 
-This MCP server (`npi-mcp`) provides a normalized interface to the NPPES NPI Registry API, allowing the CredentialWatch agent system to search for healthcare providers and retrieve detailed provider information.
+## Architecture
 
-## How it works
+This project implements a Model Context Protocol (MCP) server that acts as a bridge between an AI Agent and the NPPES NPI Registry.
 
-The server implements the Model Context Protocol (MCP) using HTTP + SSE. It exposes two tools:
+1.  **Agent (Client)**: An AI agent (e.g., LangGraph) running in a separate environment connects to this MCP server.
+2.  **MCP Server (`npi_mcp_server`)**: This server, running in a Hugging Face Space, exposes tools defined in the MCP specification (`search_npi_providers`, `get_npi_provider`).
+3.  **Modal NPI API (`NPI_API`)**: The MCP server does not call the NPPES Registry directly. Instead, it forwards requests to a Modal-hosted FastAPI service (`NPI_API`). This Modal service handles the complex logic of querying the NPPES API and normalizing the data.
+4.  **NPPES Registry**: The ultimate source of truth for provider data.
 
-1. **`search_providers`**: Searches for providers using a flexible query string (handling names and organization names) along with optional filters for state and taxonomy. It aggregates results from both Individual (NPI-1) and Organization (NPI-2) searches and normalizes the output.
-2. **`get_provider_by_npi`**: Retrieves full details for a specific NPI, including all addresses and taxonomies, normalized into a clean JSON structure.
+**Flow:**
+`Agent` -> `MCP Server (this repo)` -> `Modal NPI API` -> `NPPES Registry`
 
-## Deployment
+## Tools
 
-The server is built with **FastAPI** and uses **uv** for dependency management. It is designed to be deployed as a stateless service (e.g., on Hugging Face Spaces).
+### 1. `search_npi_providers`
 
-### Endpoints
-- `/sse`: The MCP SSE endpoint for connecting agents.
-- `/messages`: The endpoint for sending JSON-RPC messages (handled via the SSE session).
-- `/healthz`: A simple health check endpoint.
+Searches for healthcare providers.
 
-## Usage
+**Arguments:**
+*   `query` (str): Name or organization name.
+*   `state` (str, optional): 2-letter state code.
+*   `taxonomy` (str, optional): Taxonomy code.
 
-Agents connect to the `/sse` endpoint to establish a session and discover tools. They can then invoke tools by sending JSON-RPC requests to the `/messages` endpoint (linked via session ID).
+**Example Invocation (JSON arguments):**
+```json
+{
+  "query": "Mayo Clinic",
+  "state": "MN"
+}
+```
+
+### 2. `get_npi_provider`
+
+Retrieves details for a specific NPI.
+
+**Arguments:**
+*   `npi` (str): 10-digit NPI number.
+
+**Example Invocation (JSON arguments):**
+```json
+{
+  "npi": "1234567890"
+}
+```

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
-name = "npi-mcp"
+name = "npi-mcp-server"
 version = "0.1.0"
-description = "MCP server for NPPES NPI Registry"
+description = "MCP server for NPPES NPI Registry, forwarding to Modal NPI API"
 requires-python = ">=3.11"
 dependencies = [
     "fastapi>=0.100.0",
@@ -9,8 +9,10 @@ dependencies = [
     "httpx>=0.24.0",
     "pydantic>=2.0.0",
     "mcp>=1.0.0",
-    "sse-starlette>=1.8.0",
-    # Dev dependencies included here for simplicity in hackathon context
+]
+
+[project.optional-dependencies]
+dev = [
     "pytest>=7.0.0",
     "pytest-asyncio>=0.21.0",
     "pytest-mock>=3.10.0",
@@ -21,4 +23,9 @@ requires = ["hatchling"]
 build-backend = "hatchling.build"
 
 [tool.hatch.build.targets.wheel]
-packages = ["src/npi_mcp"]
+packages = ["src/npi_mcp_server"]
+
+[tool.pytest.ini_options]
+pythonpath = ["src"]
+testpaths = ["tests"]
+asyncio_mode = "auto"

src/npi_mcp/main.py

@@ -1,121 +0,0 @@
-import logging
-from contextlib import asynccontextmanager
-import uuid
-
-from fastapi import FastAPI, Request
-from starlette.responses import JSONResponse
-from sse_starlette.sse import EventSourceResponse
-
-# mcp imports
-from mcp.server.sse import SseServerTransport
-from npi_mcp.mcp_tools import mcp_server, npi_client
-
-# Configure logging
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
-
-# We need to track active SSE sessions to route POST messages to the correct transport
-# In a distributed deployment, this should be in an external store (e.g. Redis).
-sse_transports = {}
-
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    # Startup
-    logger.info("Starting NPI MCP Server...")
-    yield
-    # Shutdown
-    logger.info("Shutting down NPI MCP Server...")
-    await npi_client.close()
-
-app = FastAPI(lifespan=lifespan)
-
-@app.get("/healthz")
-async def healthcheck():
-    """Health check endpoint."""
-    return {"status": "ok"}
-
-@app.get("/sse")
-async def handle_sse(request: Request):
-    """
-    Handle incoming SSE connection.
-    Creates a new SseServerTransport and runs the MCP server loop for this session.
-    """
-    session_id = str(uuid.uuid4())
-
-    # Construct the endpoint URL that the client should use for subsequent messages
-    # This URL is sent to the client in the initial 'endpoint' event.
-    # Note: request.url_for handles the base URL automatically.
-    endpoint_url = str(request.url_for("handle_messages")) + f"?session_id={session_id}"
-
-    logger.info(f"New SSE connection: {session_id}")
-
-    # Create the transport
-    transport = SseServerTransport(endpoint_url)
-
-    # Store it so handle_messages can find it
-    sse_transports[session_id] = transport
-
-    async def event_generator():
-        try:
-            # mcp_server.run connects the server logic to the transport
-            # It reads from transport.incoming_messages and writes to transport.outgoing_messages
-            # initialization_options can be passed if needed
-            async with mcp_server.run(
-                transport.read_incoming(),
-                transport.write_outgoing(),
-                initialization_options={}
-            ):
-                # The transport should yield the 'endpoint' event immediately upon connection?
-                # SseServerTransport logic typically handles sending the endpoint event at start.
-                # We just need to iterate over outgoing messages and yield them as SSE events.
-
-                async for message in transport.outgoing_messages():
-                    # message is an SSEMessage object usually, or we need to format it?
-                    # mcp.server.sse.SseServerTransport.outgoing_messages yields starlette ServerSentEvent objects or similar?
-                    # Let's assume it yields objects compatible with EventSourceResponse or we need to extract.
-
-                    # Checking `mcp` implementation (mental model):
-                    # It likely yields ServerSentEvent objects.
-                    yield message
-
-        except Exception as e:
-            logger.error(f"Error in SSE session {session_id}: {e}")
-        finally:
-            logger.info(f"Closing SSE session: {session_id}")
-            sse_transports.pop(session_id, None)
-
-    return EventSourceResponse(event_generator())
-
-@app.post("/messages")
-async def handle_messages(request: Request):
-    """
-    Handle incoming JSON-RPC messages from the client.
-    Routes the message to the correct SSE transport based on session_id.
-    """
-    session_id = request.query_params.get("session_id")
-
-    if not session_id:
-        # Some clients might pass it in the body or header? Spec says "endpoint" URI.
-        # We encoded it in the query param.
-        return JSONResponse(status_code=400, content={"error": "Missing session_id"})
-
-    if session_id not in sse_transports:
-        return JSONResponse(status_code=404, content={"error": "Session not found or expired"})
-
-    transport = sse_transports[session_id]
-
-    try:
-        # Read the JSON-RPC message
-        message = await request.json()
-    except Exception:
-        return JSONResponse(status_code=400, content={"error": "Invalid JSON"})
-
-    # Pass the message to the transport
-    # The transport puts it into the input queue which mcp_server.run consumes
-    await transport.receive_json_message(message)
-
-    return JSONResponse(content={"status": "accepted"})
-
-if __name__ == "__main__":
-    import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=8000)

src/npi_mcp/mcp_tools.py

@@ -1,61 +0,0 @@
-from typing import Any, List
-import mcp.types as types
-from mcp.server import Server
-from npi_mcp.npi_client import NPIClient
-from npi_mcp.models import SearchProvidersArgs, GetProviderArgs
-
-# Create the MCP Server instance
-mcp_server = Server("npi-mcp")
-
-# We will need a way to pass the NPIClient to the tools.
-# We can instantiate it globally or contextually.
-# For simplicity, we'll use a global client, but we need to manage its lifecycle.
-
-npi_client = NPIClient()
-
-@mcp_server.list_tools()
-async def list_tools() -> List[types.Tool]:
-    return [
-        types.Tool(
-            name="search_providers",
-            description="Search for healthcare providers in the NPI Registry by name, organization, state, or taxonomy.",
-            inputSchema=SearchProvidersArgs.model_json_schema(),
-        ),
-        types.Tool(
-            name="get_provider_by_npi",
-            description="Retrieve detailed information about a specific provider using their NPI number.",
-            inputSchema=GetProviderArgs.model_json_schema(),
-        ),
-    ]
-
-@mcp_server.call_tool()
-async def call_tool(name: str, arguments: Any) -> List[types.TextContent]:
-    if name == "search_providers":
-        # Validate arguments
-        args = SearchProvidersArgs(**arguments)
-
-        results = await npi_client.search_providers(
-            query=args.query,
-            state=args.state,
-            taxonomy=args.taxonomy
-        )
-
-        # Format as JSON string
-        json_results = [r.model_dump_json() for r in results]
-        # Or return a single JSON list
-        import json
-        final_json = json.dumps([r.model_dump() for r in results], indent=2)
-
-        return [types.TextContent(type="text", text=final_json)]
-
-    elif name == "get_provider_by_npi":
-        args = GetProviderArgs(**arguments)
-        result = await npi_client.get_provider_by_npi(args.npi)
-
-        if result:
-            return [types.TextContent(type="text", text=result.model_dump_json(indent=2))]
-        else:
-            return [types.TextContent(type="text", text=f"{{ 'error': 'Provider with NPI {args.npi} not found.' }}")]
-
-    else:
-        raise ValueError(f"Unknown tool: {name}")

src/npi_mcp/npi_client.py

@@ -1,211 +0,0 @@
-import httpx
-import logging
-from typing import List, Optional, Dict, Any
-
-from npi_mcp.models import ProviderSummary, ProviderDetail, Address, Taxonomy
-
-logger = logging.getLogger(__name__)
-
-class NPIClient:
-    BASE_URL = "https://npiregistry.cms.hhs.gov/api/"
-
-    def __init__(self):
-        self.client = httpx.AsyncClient(timeout=30.0)
-
-    async def close(self):
-        await self.client.aclose()
-
-    def _normalize_address(self, addr_data: Dict[str, Any]) -> Address:
-        """Helper to convert API address format to our Address model."""
-        return Address(
-            line1=addr_data.get("address_1", ""),
-            line2=addr_data.get("address_2") or None,
-            city=addr_data.get("city", ""),
-            state=addr_data.get("state", ""),
-            postal_code=addr_data.get("postal_code", "")[:5], # Normalize to 5 digit for simplicity? Or keep full.
-            country=addr_data.get("country_code", "US")
-        )
-
-    def _get_full_name(self, basic: Dict[str, Any], enumeration_type: str) -> str:
-        if enumeration_type == "NPI-2":
-            return basic.get("organization_name", "Unknown Organization")
-        else:
-            first = basic.get("first_name", "")
-            last = basic.get("last_name", "")
-            credential = basic.get("credential", "")
-            name = f"{first} {last}".strip()
-            if credential:
-                name += f", {credential}"
-            return name
-
-    def _extract_primary_taxonomy(self, taxonomies: List[Dict[str, Any]]) -> tuple[Optional[str], Optional[str]]:
-        """Returns (code, description) of primary taxonomy."""
-        for tax in taxonomies:
-            if tax.get("primary") is True:
-                return tax.get("code"), tax.get("desc")
-        # Fallback to first if no primary
-        if taxonomies:
-            return taxonomies[0].get("code"), taxonomies[0].get("desc")
-        return None, None
-
-    async def search_providers(
-        self,
-        query: str,
-        state: Optional[str] = None,
-        taxonomy: Optional[str] = None
-    ) -> List[ProviderSummary]:
-        """
-        Searches for providers.
-        Since the API splits fields, we try to be smart about 'query'.
-        """
-        results: List[Dict[str, Any]] = []
-
-        # Strategy:
-        # 1. Generic Organization Search (wildcard)
-        # 2. Individual Search (splitting query)
-
-        # We'll make parallel requests or sequential.
-        # API requires specific fields.
-
-        params_common = {
-            "version": "2.1",
-            "limit": 50  # Reasonable limit
-        }
-        if state:
-            params_common["state"] = state
-        if taxonomy:
-            params_common["taxonomy_description"] = taxonomy
-            # Note: API doc says "taxonomy_description", but often code works or is handled.
-            # If "207RC0000X" is passed, we rely on the API handling it in description or matching.
-            # If not, this might be a limitation.
-
-        search_requests = []
-
-        # Request 1: Organization
-        req_org = params_common.copy()
-        req_org["enumeration_type"] = "NPI-2"
-        req_org["organization_name"] = f"{query}*"
-        search_requests.append(req_org)
-
-        # Request 2: Individual (Last Name match)
-        # If query is single word
-        parts = query.split()
-        if len(parts) == 1:
-            req_ind = params_common.copy()
-            req_ind["enumeration_type"] = "NPI-1"
-            req_ind["last_name"] = f"{query}*"
-            search_requests.append(req_ind)
-        elif len(parts) >= 2:
-            # First Last
-            req_ind = params_common.copy()
-            req_ind["enumeration_type"] = "NPI-1"
-            req_ind["first_name"] = parts[0]
-            req_ind["last_name"] = f"{parts[-1]}*" # Use wildcard on last name
-            search_requests.append(req_ind)
-
-        # Execute requests
-        # We run them sequentially for simplicity in this implementation,
-        # but could use asyncio.gather
-
-        seen_npis = set()
-        normalized_results = []
-
-        for params in search_requests:
-            try:
-                resp = await self.client.get(self.BASE_URL, params=params)
-                resp.raise_for_status()
-                data = resp.json()
-
-                # API returns { "result_count": ..., "results": [...] } or errors
-                items = data.get("results", [])
-
-                for item in items:
-                    npi = item.get("number")
-                    if npi in seen_npis:
-                        continue
-                    seen_npis.add(npi)
-
-                    basic = item.get("basic", {})
-                    enum_type = item.get("enumeration_type", "UNKNOWN")
-                    # Map NPI-1 to INDIVIDUAL, NPI-2 to ORGANIZATION
-                    type_str = "INDIVIDUAL" if enum_type == "NPI-1" else "ORGANIZATION"
-
-                    full_name = self._get_full_name(basic, enum_type)
-
-                    taxonomies = item.get("taxonomies", [])
-                    prim_code, prim_desc = self._extract_primary_taxonomy(taxonomies)
-
-                    # Find primary address (usually location address)
-                    addresses = item.get("addresses", [])
-                    primary_addr_data = next(
-                        (a for a in addresses if a.get("address_purpose") == "LOCATION"),
-                        addresses[0] if addresses else {}
-                    )
-
-                    normalized_results.append(ProviderSummary(
-                        npi=str(npi),
-                        full_name=full_name,
-                        enumeration_type=type_str,
-                        primary_taxonomy=prim_code,
-                        primary_specialty=prim_desc,
-                        primary_address=self._normalize_address(primary_addr_data)
-                    ))
-            except Exception as e:
-                logger.error(f"Error querying NPI API with params {params}: {e}")
-                # Continue to next request strategy
-                continue
-
-        return normalized_results
-
-    async def get_provider_by_npi(self, npi: str) -> Optional[ProviderDetail]:
-        params = {
-            "version": "2.1",
-            "number": npi
-        }
-        try:
-            resp = await self.client.get(self.BASE_URL, params=params)
-            resp.raise_for_status()
-            data = resp.json()
-
-            results = data.get("results", [])
-            if not results:
-                return None
-
-            item = results[0]
-            basic = item.get("basic", {})
-            enum_type = item.get("enumeration_type", "UNKNOWN")
-            type_str = "INDIVIDUAL" if enum_type == "NPI-1" else "ORGANIZATION"
-
-            full_name = self._get_full_name(basic, enum_type)
-
-            # Addresses
-            raw_addresses = item.get("addresses", [])
-            addresses = [self._normalize_address(a) for a in raw_addresses]
-
-            # Taxonomies
-            raw_taxonomies = item.get("taxonomies", [])
-            taxonomies = []
-            for t in raw_taxonomies:
-                taxonomies.append(Taxonomy(
-                    code=t.get("code", ""),
-                    description=t.get("desc"),
-                    primary=t.get("primary", False),
-                    state=t.get("state"),
-                    license=t.get("license")
-                ))
-
-            return ProviderDetail(
-                npi=str(item.get("number")),
-                full_name=full_name,
-                enumeration_type=type_str,
-                addresses=addresses,
-                taxonomies=taxonomies
-            )
-
-        except httpx.HTTPStatusError as e:
-            if e.response.status_code == 404:
-                return None
-            raise e
-        except Exception as e:
-            logger.error(f"Error fetching NPI {npi}: {e}")
-            raise e

src/npi_mcp_server/__init__.py

@@ -0,0 +1,5 @@
+# Expose main entry point
+from npi_mcp_server.main import mcp
+
+def main():
+    mcp.run()

src/npi_mcp_server/config.py

@@ -0,0 +1,5 @@
+import os
+
+# Base URL for the Modal NPI API service
+# Default to localhost for testing if not set, but in prod it should be set.
+NPI_API_BASE_URL = os.environ.get("NPI_API_BASE_URL", "http://localhost:8000")

src/npi_mcp_server/main.py

@@ -0,0 +1,37 @@
+from mcp.server.fastmcp import FastMCP
+from npi_mcp_server.tools import search_providers as _search_providers, get_provider_by_npi as _get_provider_by_npi
+from npi_mcp_server.schemas import ProviderSummary, ProviderDetail
+
+# Initialize FastMCP server
+# "npi-mcp" is the name of the server
+mcp = FastMCP("npi-mcp")
+
+@mcp.tool()
+async def search_providers(query: str, state: str | None = None, taxonomy: str | None = None) -> list[ProviderSummary]:
+    """
+    Search for healthcare providers in the NPI Registry by name, organization, state, or taxonomy.
+    Fowards requests to the NPI API service.
+
+    Args:
+        query: Name of the provider (first/last) or organization, or a generic search term.
+        state: 2-letter state code (e.g. 'CA', 'NY').
+        taxonomy: Taxonomy code or description (e.g. '207RC0000X').
+    """
+    return await _search_providers(query, state, taxonomy)
+
+@mcp.tool()
+async def get_provider_by_npi(npi: str) -> ProviderDetail | str:
+    """
+    Retrieve detailed information about a specific provider using their NPI number.
+
+    Args:
+        npi: The 10-digit NPI number.
+    """
+    result = await _get_provider_by_npi(npi)
+    if result:
+        return result
+    return f"Provider with NPI {npi} not found."
+
+# Entry point for running the server directly (e.g. for testing)
+if __name__ == "__main__":
+    mcp.run()

src/{npi_mcp/models.py → npi_mcp_server/schemas.py}

@@ -11,7 +11,8 @@ class SearchProvidersArgs(BaseModel):
 class GetProviderArgs(BaseModel):
     npi: str = Field(..., description="The 10-digit NPI number.")
 
-# --- Normalized Response Models ---
+# --- NPI API Response Models ---
+# These mirror the structure returned by the Modal NPI_API.
 
 class Address(BaseModel):
     line1: str
@@ -29,6 +30,9 @@ class ProviderSummary(BaseModel):
     primary_specialty: Optional[str] = None
     primary_address: Address
 
+class SearchProvidersResponse(BaseModel):
+    results: List[ProviderSummary]
+
 class Taxonomy(BaseModel):
     code: str
     description: Optional[str] = None

src/npi_mcp_server/tools.py

@@ -0,0 +1,86 @@
+import httpx
+import logging
+from typing import List, Optional
+
+from npi_mcp_server.config import NPI_API_BASE_URL
+from npi_mcp_server.schemas import (
+    ProviderSummary,
+    ProviderDetail,
+    SearchProvidersResponse
+)
+
+logger = logging.getLogger(__name__)
+
+async def search_providers(
+    query: str,
+    state: Optional[str] = None,
+    taxonomy: Optional[str] = None
+) -> List[ProviderSummary]:
+    """
+    Search for healthcare providers via the NPI API.
+
+    Args:
+        query: Name (first/last) or organization name.
+        state: Two-letter state code.
+        taxonomy: Taxonomy code or description.
+
+    Returns:
+        List of ProviderSummary objects.
+    """
+    url = f"{NPI_API_BASE_URL.rstrip('/')}/search_providers"
+
+    payload = {
+        "query": query,
+        "state": state,
+        "taxonomy": taxonomy
+    }
+
+    # Remove None values
+    payload = {k: v for k, v in payload.items() if v is not None}
+
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.post(url, json=payload, timeout=30.0)
+            response.raise_for_status()
+
+            data = response.json()
+            # Expecting: { "results": [ ... ] }
+            return SearchProvidersResponse(**data).results
+
+        except httpx.HTTPStatusError as e:
+            logger.error(f"HTTP error calling NPI API: {e.response.status_code} - {e.response.text}")
+            raise RuntimeError(f"NPI API returned error: {e.response.status_code}") from e
+        except Exception as e:
+            logger.error(f"Error calling NPI API: {e}")
+            raise RuntimeError(f"Failed to search providers: {str(e)}") from e
+
+async def get_provider_by_npi(npi: str) -> Optional[ProviderDetail]:
+    """
+    Retrieve details for a specific provider by NPI.
+
+    Args:
+        npi: 10-digit NPI string.
+
+    Returns:
+        ProviderDetail object or None if not found.
+    """
+    url = f"{NPI_API_BASE_URL.rstrip('/')}/provider/{npi}"
+
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.get(url, timeout=30.0)
+
+            if response.status_code == 404:
+                return None
+
+            response.raise_for_status()
+
+            data = response.json()
+            return ProviderDetail(**data)
+
+        except httpx.HTTPStatusError as e:
+            logger.error(f"HTTP error calling NPI API: {e.response.status_code} - {e.response.text}")
+            raise RuntimeError(f"NPI API returned error: {e.response.status_code}") from e
+        except Exception as e:
+            logger.error(f"Error calling NPI API: {e}")
+            raise RuntimeError(f"Failed to get provider: {str(e)}") from e

tests/test_npi_mcp.py

@@ -1,112 +0,0 @@
-import pytest
-from httpx import Response
-from npi_mcp.npi_client import NPIClient
-from npi_mcp.models import ProviderSummary, ProviderDetail
-
-# Mock data
-MOCK_SEARCH_RESPONSE_IND = {
-    "result_count": 1,
-    "results": [
-        {
-            "number": "1234567890",
-            "basic": {
-                "first_name": "John",
-                "last_name": "Doe",
-                "credential": "MD"
-            },
-            "enumeration_type": "NPI-1",
-            "taxonomies": [
-                {"code": "207RC0000X", "desc": "Cardiology", "primary": True}
-            ],
-            "addresses": [
-                {
-                    "address_purpose": "LOCATION",
-                    "address_1": "123 Main St",
-                    "city": "Anytown",
-                    "state": "CA",
-                    "postal_code": "90210",
-                    "country_code": "US"
-                }
-            ]
-        }
-    ]
-}
-
-MOCK_SEARCH_RESPONSE_ORG = {
-    "result_count": 1,
-    "results": [
-        {
-            "number": "9876543210",
-            "basic": {
-                "organization_name": "General Hospital"
-            },
-            "enumeration_type": "NPI-2",
-            "taxonomies": [],
-            "addresses": [
-                {
-                    "address_purpose": "LOCATION",
-                    "address_1": "456 Health Blvd",
-                    "city": "Metropolis",
-                    "state": "NY",
-                    "postal_code": "10001",
-                    "country_code": "US"
-                }
-            ]
-        }
-    ]
-}
-
-import httpx
-
-@pytest.mark.asyncio
-async def test_search_providers_individual(mocker):
-    # Mock httpx client
-    # Note: raise_for_status requires a request object
-    resp = Response(200, json=MOCK_SEARCH_RESPONSE_IND)
-    resp._request = httpx.Request("GET", "https://mock")
-    mock_get = mocker.patch("httpx.AsyncClient.get", return_value=resp)
-
-    client = NPIClient()
-    results = await client.search_providers(query="John Doe")
-
-    assert len(results) >= 1
-    p = results[0]
-    assert p.full_name == "John Doe, MD"
-    assert p.enumeration_type == "INDIVIDUAL"
-    assert p.primary_address.city == "Anytown"
-
-    await client.close()
-
-@pytest.mark.asyncio
-async def test_search_providers_org(mocker):
-    # Mock httpx client
-    resp = Response(200, json=MOCK_SEARCH_RESPONSE_ORG)
-    resp._request = httpx.Request("GET", "https://mock")
-    mock_get = mocker.patch("httpx.AsyncClient.get", return_value=resp)
-
-    client = NPIClient()
-    results = await client.search_providers(query="General Hospital")
-
-    assert len(results) >= 1
-    p = results[0]
-    assert p.full_name == "General Hospital"
-    assert p.enumeration_type == "ORGANIZATION"
-
-    await client.close()
-
-@pytest.mark.asyncio
-async def test_get_provider_by_npi(mocker):
-    resp = Response(200, json=MOCK_SEARCH_RESPONSE_IND)
-    resp._request = httpx.Request("GET", "https://mock")
-    mock_get = mocker.patch("httpx.AsyncClient.get", return_value=resp)
-
-    client = NPIClient()
-    result = await client.get_provider_by_npi("1234567890")
-
-    assert result is not None
-    assert result.npi == "1234567890"
-    assert result.full_name == "John Doe, MD"
-    assert len(result.taxonomies) == 1
-    assert result.taxonomies[0].code == "207RC0000X"
-
-    await client.close()

tests/test_npi_mcp_server.py

@@ -0,0 +1,95 @@
+import pytest
+from httpx import Response
+import httpx
+from npi_mcp_server.tools import search_providers, get_provider_by_npi
+from npi_mcp_server.schemas import ProviderSummary, ProviderDetail
+
+# Mock data matching the Modal NPI API contract
+MOCK_SEARCH_RESPONSE = {
+    "results": [
+        {
+            "npi": "1234567890",
+            "full_name": "John Doe, MD",
+            "enumeration_type": "INDIVIDUAL",
+            "primary_taxonomy": "207RC0000X",
+            "primary_specialty": "Cardiology",
+            "primary_address": {
+                "line1": "123 Main St",
+                "city": "Anytown",
+                "state": "CA",
+                "postal_code": "90210",
+                "country": "US"
+            }
+        }
+    ]
+}
+
+MOCK_PROVIDER_DETAIL = {
+    "npi": "1234567890",
+    "full_name": "John Doe, MD",
+    "enumeration_type": "INDIVIDUAL",
+    "addresses": [
+        {
+            "line1": "123 Main St",
+            "city": "Anytown",
+            "state": "CA",
+            "postal_code": "90210",
+            "country": "US"
+        }
+    ],
+    "taxonomies": [
+        {
+            "code": "207RC0000X",
+            "description": "Cardiology",
+            "primary": True,
+            "state": "CA",
+            "license": "12345"
+        }
+    ]
+}
+
+@pytest.mark.asyncio
+async def test_search_providers(mocker):
+    # Mock httpx client post
+    resp = Response(200, json=MOCK_SEARCH_RESPONSE)
+    resp._request = httpx.Request("POST", "https://mock/search_providers")
+
+    mock_post = mocker.patch("httpx.AsyncClient.post", return_value=resp)
+
+    results = await search_providers(query="John Doe")
+
+    assert len(results) == 1
+    p = results[0]
+    assert p.npi == "1234567890"
+    assert p.full_name == "John Doe, MD"
+
+    # Check if correct URL and payload were sent
+    # We can check mock_post.call_args
+    call_args = mock_post.call_args
+    assert "search_providers" in call_args[0][0]
+    assert call_args[1]['json'] == {"query": "John Doe"}
+
+@pytest.mark.asyncio
+async def test_get_provider_by_npi(mocker):
+    # Mock httpx client get
+    resp = Response(200, json=MOCK_PROVIDER_DETAIL)
+    resp._request = httpx.Request("GET", "https://mock/provider/1234567890")
+
+    mock_get = mocker.patch("httpx.AsyncClient.get", return_value=resp)
+
+    result = await get_provider_by_npi("1234567890")
+
+    assert result is not None
+    assert result.npi == "1234567890"
+    assert result.full_name == "John Doe, MD"
+    assert len(result.taxonomies) == 1
+
+@pytest.mark.asyncio
+async def test_get_provider_by_npi_not_found(mocker):
+    resp = Response(404)
+    resp._request = httpx.Request("GET", "https://mock/provider/0000000000")
+
+    mock_get = mocker.patch("httpx.AsyncClient.get", return_value=resp)
+
+    result = await get_provider_by_npi("0000000000")
+    assert result is None

uv.lock

@@ -336,7 +336,7 @@ wheels = [
 ]
 
 [[package]]
-name = "npi-mcp"
+name = "npi-mcp-server"
 version = "0.1.0"
 source = { editable = "." }
 dependencies = [
@@ -344,11 +344,14 @@ dependencies = [
     { name = "httpx" },
     { name = "mcp" },
     { name = "pydantic" },
+    { name = "uvicorn" },
+]
+
+[package.optional-dependencies]
+dev = [
     { name = "pytest" },
     { name = "pytest-asyncio" },
     { name = "pytest-mock" },
-    { name = "sse-starlette" },
-    { name = "uvicorn" },
 ]
 
 [package.metadata]
@@ -357,12 +360,12 @@ requires-dist = [
     { name = "httpx", specifier = ">=0.24.0" },
     { name = "mcp", specifier = ">=1.0.0" },
     { name = "pydantic", specifier = ">=2.0.0" },
-    { name = "pytest", specifier = ">=7.0.0" },
-    { name = "pytest-asyncio", specifier = ">=0.21.0" },
-    { name = "pytest-mock", specifier = ">=3.10.0" },
-    { name = "sse-starlette", specifier = ">=1.8.0" },
+    { name = "pytest", marker = "extra == 'dev'", specifier = ">=7.0.0" },
+    { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=0.21.0" },
+    { name = "pytest-mock", marker = "extra == 'dev'", specifier = ">=3.10.0" },
     { name = "uvicorn", specifier = ">=0.20.0" },
 ]
+provides-extras = ["dev"]
 
 [[package]]
 name = "packaging"