Spaces:
Sleeping
Sleeping
File size: 2,989 Bytes
f8c07f6 1cc51a2 ca50165 f8c07f6 ca50165 f8c07f6 ca50165 f8c07f6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 |
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}
logger.info(f"Searching providers with payload: {payload}")
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
}
async with httpx.AsyncClient() as client:
try:
response = await client.post(url, json=payload, headers=headers, timeout=30.0)
response.raise_for_status()
data = response.json()
logger.info(f"Received response from NPI API: {data}")
# 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
|