Merge pull request #1 from humandotlearning/jules/cred-db-mcp-server

pyproject.toml

@@ -1,5 +1,5 @@
 [project]
-name = "cred_db_mcp"
+name = "cred_db_mcp_server"
 version = "0.1.0"
 description = "Credential Database MCP Server"
 requires-python = ">=3.11"
@@ -9,7 +9,8 @@ dependencies = [
     "sqlalchemy>=2.0.0",
     "pydantic>=2.0.0",
     "httpx>=0.24.0",
-    "python-dotenv>=1.0.0"
+    "python-dotenv>=1.0.0",
+    "gradio>=5.0.0",
 ]
 
 [project.optional-dependencies]
@@ -22,6 +23,9 @@ test = [
 requires = ["hatchling"]
 build-backend = "hatchling.build"
 
+[tool.hatch.build.targets.wheel]
+packages = ["src/cred_db_mcp_server"]
+
 [tool.pytest.ini_options]
 pythonpath = "src"
 testpaths = ["tests"]

src/cred_db_mcp/db.py

@@ -1,33 +0,0 @@
-import os
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker, DeclarativeBase
-
-# Default to a local file for dev/test if not specified
-DB_PATH = os.getenv("DB_PATH", "credentialwatch.db")
-DATABASE_URL = f"sqlite:///{DB_PATH}"
-
-# specific check for checks that might run in different environments
-if DB_PATH == ":memory:":
-    DATABASE_URL = "sqlite:///:memory:"
-
-engine = create_engine(
-    DATABASE_URL,
-    connect_args={"check_same_thread": False}  # Needed for SQLite
-)
-
-SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
-
-class Base(DeclarativeBase):
-    pass
-
-def get_db():
-    """Dependency for FastAPI to get a DB session."""
-    db = SessionLocal()
-    try:
-        yield db
-    finally:
-        db.close()
-
-def init_db():
-    """Helper to initialize the DB (create tables)."""
-    Base.metadata.create_all(bind=engine)

src/cred_db_mcp/main.py

@@ -1,98 +0,0 @@
-from fastapi import FastAPI, Depends, HTTPException, Body
-from sqlalchemy.orm import Session
-from typing import List, Union
-
-from .db import get_db, init_db
-from .mcp_tools import MCPTools
-from .schemas import (
-    SyncProviderInput, AddCredentialInput, ListExpiringInput, GetSnapshotInput,
-    ProviderRead, CredentialRead, ExpiringCredentialItem, ProviderSnapshot, ToolError
-)
-
-app = FastAPI(title="CredentialWatch DB MCP", version="0.1.0")
-
-# Initialize DB on startup (for simple deployments)
-@app.on_event("startup")
-def on_startup():
-    init_db()
-
-def get_mcp_tools(db: Session = Depends(get_db)) -> MCPTools:
-    return MCPTools(db)
-
-# --- MCP Tool Endpoints ---
-# Pattern: /mcp/tools/{tool_name}
-# We use POST for all of them as they are "function calls"
-
-@app.post("/mcp/tools/sync_provider_from_npi", response_model=Union[ProviderRead, ToolError])
-async def sync_provider_from_npi(
-    args: SyncProviderInput,
-    tools: MCPTools = Depends(get_mcp_tools)
-):
-    """
-    Syncs provider data from the NPI Registry via npi_mcp.
-    """
-    result = await tools.sync_provider_from_npi(args.npi)
-    if isinstance(result, dict) and "error" in result:
-        # Return generic error structure instead of 400 for agent handling if preferred,
-        # but usually 400 is better for HTTP.
-        # The prompt says "return a structured error".
-        return ToolError(error=result["error"])
-    return result
-
-@app.post("/mcp/tools/add_or_update_credential", response_model=Union[CredentialRead, ToolError])
-def add_or_update_credential(
-    args: AddCredentialInput,
-    tools: MCPTools = Depends(get_mcp_tools)
-):
-    """
-    Adds or updates a credential record.
-    """
-    result = tools.add_or_update_credential(
-        provider_id=args.provider_id,
-        type=args.type,
-        issuer=args.issuer,
-        number=args.number,
-        expiry_date=args.expiry_date
-    )
-    if isinstance(result, dict) and "error" in result:
-        return ToolError(error=result["error"])
-    return result
-
-@app.post("/mcp/tools/list_expiring_credentials", response_model=List[ExpiringCredentialItem])
-def list_expiring_credentials(
-    args: ListExpiringInput,
-    tools: MCPTools = Depends(get_mcp_tools)
-):
-    """
-    Lists credentials expiring within a window.
-    """
-    return tools.list_expiring_credentials(
-        window_days=args.window_days,
-        dept=args.dept,
-        location=args.location
-    )
-
-@app.post("/mcp/tools/get_provider_snapshot", response_model=Union[ProviderSnapshot, ToolError])
-def get_provider_snapshot(
-    args: GetSnapshotInput,
-    tools: MCPTools = Depends(get_mcp_tools)
-):
-    """
-    Gets a full snapshot of a provider.
-    """
-    result = tools.get_provider_snapshot(
-        provider_id=args.provider_id,
-        npi=args.npi
-    )
-    if isinstance(result, dict) and "error" in result:
-        return ToolError(error=result["error"])
-    return result
-
-# Simple info endpoint
-@app.get("/")
-def read_root():
-    return {
-        "service": "cred_db_mcp",
-        "status": "running",
-        "docs": "/docs"
-    }

src/cred_db_mcp/mcp_tools.py

@@ -1,185 +0,0 @@
-from sqlalchemy.orm import Session
-from sqlalchemy import select, and_, func
-from datetime import date, timedelta, datetime
-import json
-
-from .models import Provider, Credential, Alert
-from .schemas import (
-    ProviderRead, CredentialRead, AlertRead,
-    ExpiringCredentialItem, ProviderSnapshot
-)
-from .npi_client import NPIClient
-
-class MCPTools:
-    def __init__(self, db: Session):
-        self.db = db
-        self.npi_client = NPIClient()
-
-    async def sync_provider_from_npi(self, npi: str):
-        # 1. Call npi_mcp
-        npi_data = await self.npi_client.get_provider_by_npi(npi)
-
-        if not npi_data:
-            return {"error": f"NPI {npi} not found in upstream registry"}
-
-        # Map NPI data to our schema.
-        # Assuming npi_data has keys: npi, first_name, last_name, taxonomy_desc, practice_address
-        # This mapping is hypothetical based on typical NPI registry fields.
-
-        full_name = f"{npi_data.get('first_name', '')} {npi_data.get('last_name', '')}".strip()
-        if not full_name:
-            full_name = npi_data.get('organization_name', 'Unknown')
-
-        specialty = npi_data.get('taxonomy_desc', 'Unknown')
-        address_obj = npi_data.get('practice_address', {})
-        # Simple string representation of location
-        location = f"{address_obj.get('city', '')}, {address_obj.get('state', '')}" if isinstance(address_obj, dict) else str(address_obj)
-
-        # 2. Update or Create in DB
-        provider = self.db.scalar(select(Provider).where(Provider.npi == npi))
-
-        if not provider:
-            provider = Provider(
-                npi=npi,
-                full_name=full_name,
-                primary_specialty=specialty,
-                location=location,
-                is_active=True
-            )
-            self.db.add(provider)
-        else:
-            provider.full_name = full_name
-            provider.primary_specialty = specialty
-            provider.location = location
-            # Don't overwrite dept if set locally? assuming upstream doesn't have dept.
-
-        self.db.commit()
-        self.db.refresh(provider)
-
-        return ProviderRead.model_validate(provider)
-
-    def add_or_update_credential(
-        self, provider_id: int, type: str, issuer: str, number: str, expiry_date: str
-    ):
-        # Parse expiry_date (assuming YYYY-MM-DD)
-        try:
-            exp_date = datetime.strptime(expiry_date, "%Y-%m-%d").date()
-        except ValueError:
-             return {"error": "Invalid date format. Use YYYY-MM-DD"}
-
-        # Check provider exists
-        provider = self.db.get(Provider, provider_id)
-        if not provider:
-            return {"error": f"Provider with ID {provider_id} not found"}
-
-        # Find existing credential
-        stmt = select(Credential).where(
-            and_(
-                Credential.provider_id == provider_id,
-                Credential.type == type,
-                Credential.issuer == issuer,
-                Credential.number == number
-            )
-        )
-        credential = self.db.scalar(stmt)
-
-        if credential:
-            credential.expiry_date = exp_date
-            # Heuristic status update
-            credential.status = "active" if exp_date > date.today() else "expired"
-            credential.updated_at = datetime.now()
-        else:
-            credential = Credential(
-                provider_id=provider_id,
-                type=type,
-                issuer=issuer,
-                number=number,
-                expiry_date=exp_date,
-                status="active" if exp_date > date.today() else "expired",
-                created_at=datetime.now()
-            )
-            self.db.add(credential)
-
-        self.db.commit()
-        self.db.refresh(credential)
-
-        return CredentialRead.model_validate(credential)
-
-    def list_expiring_credentials(
-        self, window_days: int, dept: str | None = None, location: str | None = None
-    ):
-        today = date.today()
-        cutoff = today + timedelta(days=window_days)
-
-        # Build Query
-        stmt = select(Credential, Provider).join(Provider).where(
-            and_(
-                Credential.status == "active",
-                Credential.expiry_date <= cutoff,
-                Credential.expiry_date >= today # Only future or today (past would be expired)
-            )
-        )
-
-        if dept:
-            stmt = stmt.where(Provider.dept == dept)
-        if location:
-            stmt = stmt.where(Provider.location.ilike(f"%{location}%"))
-
-        results = self.db.execute(stmt).all()
-
-        output = []
-        for cred, prov in results:
-            if not cred.expiry_date:
-                continue
-
-            days_to_expiry = (cred.expiry_date - today).days
-
-            # Risk Score Heuristic
-            # 3 for <30 days, 2 for 30–60, 1 for 60–90
-            if days_to_expiry < 30:
-                risk_score = 3
-            elif days_to_expiry < 60:
-                risk_score = 2
-            else:
-                risk_score = 1
-
-            output.append(ExpiringCredentialItem(
-                provider=ProviderRead.model_validate(prov),
-                credential=CredentialRead.model_validate(cred),
-                days_to_expiry=days_to_expiry,
-                risk_score=risk_score
-            ))
-
-        return output
-
-    def get_provider_snapshot(
-        self, provider_id: int | None = None, npi: str | None = None
-    ):
-        if not provider_id and not npi:
-            return {"error": "Must provide either provider_id or npi"}
-
-        stmt = select(Provider)
-        if provider_id:
-            stmt = stmt.where(Provider.id == provider_id)
-        else:
-            stmt = stmt.where(Provider.npi == npi)
-
-        provider = self.db.scalar(stmt)
-        if not provider:
-            return {"error": "Provider not found"}
-
-        # Get Credentials
-        creds = self.db.scalars(
-            select(Credential).where(Credential.provider_id == provider.id)
-        ).all()
-
-        # Get Alerts (Optional)
-        alerts = self.db.scalars(
-            select(Alert).where(Alert.provider_id == provider.id)
-        ).all()
-
-        return ProviderSnapshot(
-            provider=ProviderRead.model_validate(provider),
-            credentials=[CredentialRead.model_validate(c) for c in creds],
-            alerts=[AlertRead.model_validate(a) for a in alerts]
-        )

src/cred_db_mcp/models.py

@@ -1,62 +0,0 @@
-from datetime import datetime, date
-from typing import Optional, List, Any
-from sqlalchemy import (
-    String, Integer, Boolean, DateTime, Date, ForeignKey, JSON, func
-)
-from sqlalchemy.orm import Mapped, mapped_column, relationship
-
-from .db import Base
-
-class Provider(Base):
-    __tablename__ = "providers"
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
-    npi: Mapped[Optional[str]] = mapped_column(String, index=True, nullable=True)
-    full_name: Mapped[str] = mapped_column(String)
-    dept: Mapped[Optional[str]] = mapped_column(String, nullable=True)
-    location: Mapped[Optional[str]] = mapped_column(String, nullable=True)
-    primary_specialty: Mapped[Optional[str]] = mapped_column(String, nullable=True)
-    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
-
-    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime, server_default=func.now(), onupdate=func.now()
-    )
-
-    credentials: Mapped[List["Credential"]] = relationship(
-        "Credential", back_populates="provider", cascade="all, delete-orphan"
-    )
-
-class Credential(Base):
-    __tablename__ = "credentials"
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
-    provider_id: Mapped[int] = mapped_column(
-        Integer, ForeignKey("providers.id"), nullable=False
-    )
-    type: Mapped[str] = mapped_column(String)  # e.g. "state_license", "board_cert"
-    issuer: Mapped[str] = mapped_column(String)
-    number: Mapped[str] = mapped_column(String)
-    status: Mapped[str] = mapped_column(String)  # "active", "expired", etc.
-
-    issue_date: Mapped[Optional[date]] = mapped_column(Date, nullable=True)
-    expiry_date: Mapped[Optional[date]] = mapped_column(Date, nullable=True)
-    last_verified_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
-
-    metadata_json: Mapped[Optional[Any]] = mapped_column(JSON, nullable=True)
-
-    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime, server_default=func.now(), onupdate=func.now()
-    )
-
-    provider: Mapped["Provider"] = relationship("Provider", back_populates="credentials")
-
-class Alert(Base):
-    __tablename__ = "alerts"
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
-    provider_id: Mapped[int] = mapped_column(Integer, ForeignKey("providers.id"))
-    message: Mapped[str] = mapped_column(String)
-    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
-    # Minimal schema for alerts as requested

src/cred_db_mcp/npi_client.py

@@ -1,40 +0,0 @@
-import os
-import httpx
-from typing import Optional, Dict, Any
-
-class NPIClient:
-    def __init__(self, base_url: Optional[str] = None):
-        self.base_url = base_url or os.getenv("NPI_MCP_URL", "http://localhost:8001")
-        # Ensure no trailing slash
-        self.base_url = self.base_url.rstrip("/")
-
-    async def get_provider_by_npi(self, npi: str) -> Optional[Dict[str, Any]]:
-        """
-        Calls the npi_mcp server to get provider details.
-        Assumes npi_mcp exposes a tool 'get_provider_by_npi' via HTTP.
-
-        The NPI MCP is expected to have a similar structure: POST /mcp/tools/get_provider_by_npi
-        """
-        url = f"{self.base_url}/mcp/tools/get_provider_by_npi"
-
-        # We assume the NPI MCP accepts arguments in the body
-        payload = {"npi": npi}
-
-        try:
-            async with httpx.AsyncClient() as client:
-                response = await client.post(url, json=payload, timeout=10.0)
-
-                if response.status_code == 200:
-                    data = response.json()
-                    # If the tool wraps return value, e.g. {"result": ...} adjust here.
-                    # For now assuming direct return or generic tool response.
-                    return data
-                elif response.status_code == 404:
-                    return None
-                else:
-                    # Log error or raise
-                    print(f"Error calling NPI MCP: {response.status_code} {response.text}")
-                    return None
-        except httpx.RequestError as e:
-             print(f"Request error calling NPI MCP: {e}")
-             return None

src/cred_db_mcp/schemas.py

@@ -1,79 +0,0 @@
-from pydantic import BaseModel, ConfigDict
-from typing import Optional, List, Any
-from datetime import date, datetime
-
-# --- Base Models ---
-
-class ProviderBase(BaseModel):
-    npi: Optional[str] = None
-    full_name: str
-    dept: Optional[str] = None
-    location: Optional[str] = None
-    primary_specialty: Optional[str] = None
-    is_active: bool = True
-
-class ProviderRead(ProviderBase):
-    id: int
-    created_at: datetime
-    updated_at: datetime
-    model_config = ConfigDict(from_attributes=True)
-
-class CredentialBase(BaseModel):
-    type: str
-    issuer: str
-    number: str
-    status: str
-    issue_date: Optional[date] = None
-    expiry_date: Optional[date] = None
-    last_verified_at: Optional[datetime] = None
-    metadata_json: Optional[Any] = None
-
-class CredentialRead(CredentialBase):
-    id: int
-    provider_id: int
-    created_at: datetime
-    updated_at: datetime
-    model_config = ConfigDict(from_attributes=True)
-
-class AlertRead(BaseModel):
-    id: int
-    provider_id: int
-    message: str
-    created_at: datetime
-    model_config = ConfigDict(from_attributes=True)
-
-# --- Tool IO Models ---
-
-class SyncProviderInput(BaseModel):
-    npi: str
-
-class AddCredentialInput(BaseModel):
-    provider_id: int
-    type: str
-    issuer: str
-    number: str
-    expiry_date: str # Expecting YYYY-MM-DD string as per prompt, or we can parse it
-
-class ListExpiringInput(BaseModel):
-    window_days: int
-    dept: Optional[str] = None
-    location: Optional[str] = None
-
-class ExpiringCredentialItem(BaseModel):
-    provider: ProviderRead
-    credential: CredentialRead
-    days_to_expiry: int
-    risk_score: int
-
-class GetSnapshotInput(BaseModel):
-    provider_id: Optional[int] = None
-    npi: Optional[str] = None
-
-class ProviderSnapshot(BaseModel):
-    provider: ProviderRead
-    credentials: List[CredentialRead]
-    alerts: Optional[List[AlertRead]] = []
-
-class ToolError(BaseModel):
-    error: str
-    details: Optional[str] = None

src/cred_db_mcp_server/__init__.py

@@ -0,0 +1,4 @@
+"""
+The `cred_db_mcp_server` package exposes an MCP server wrapper around the `CRED_API` service.
+It uses Gradio to provide the MCP endpoints.
+"""

src/cred_db_mcp_server/config.py

@@ -0,0 +1,6 @@
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+CRED_API_BASE_URL = os.getenv("CRED_API_BASE_URL", "http://localhost:8000")

src/cred_db_mcp_server/main.py

@@ -0,0 +1,75 @@
+import gradio as gr
+from .tools import (
+    sync_provider_from_npi,
+    add_or_update_credential,
+    list_expiring_credentials,
+    get_provider_snapshot
+)
+
+def main():
+    """
+    Main entry point for the Gradio MCP server.
+    """
+    # Create the Gradio interface
+    # We use a TabbedInterface to organize the tools if accessed via UI,
+    # but primarily they are exposed as MCP tools.
+
+    # Tool 1: sync_provider_from_npi
+    iface_sync = gr.Interface(
+        fn=sync_provider_from_npi,
+        inputs=[gr.Textbox(label="NPI")],
+        outputs=gr.JSON(label="Provider Data"),
+        description="Syncs a provider's data from the NPI registry.",
+        allow_flagging="never"
+    )
+
+    # Tool 2: add_or_update_credential
+    iface_add_cred = gr.Interface(
+        fn=add_or_update_credential,
+        inputs=[
+            gr.Number(label="Provider ID", precision=0),
+            gr.Textbox(label="Type"),
+            gr.Textbox(label="Issuer"),
+            gr.Textbox(label="Number"),
+            gr.Textbox(label="Expiry Date (YYYY-MM-DD)"),
+        ],
+        outputs=gr.JSON(label="Credential Data"),
+        description="Adds or updates a credential for a provider.",
+        allow_flagging="never"
+    )
+
+    # Tool 3: list_expiring_credentials
+    iface_expiring = gr.Interface(
+        fn=list_expiring_credentials,
+        inputs=[
+            gr.Number(label="Window Days", precision=0),
+            gr.Textbox(label="Department", value=None),
+            gr.Textbox(label="Location", value=None),
+        ],
+        outputs=gr.JSON(label="Expiring Credentials"),
+        description="Lists credentials expiring within a certain number of days.",
+        allow_flagging="never"
+    )
+
+    # Tool 4: get_provider_snapshot
+    iface_snapshot = gr.Interface(
+        fn=get_provider_snapshot,
+        inputs=[
+            gr.Number(label="Provider ID", precision=0, value=None),
+            gr.Textbox(label="NPI", value=None),
+        ],
+        outputs=gr.JSON(label="Provider Snapshot"),
+        description="Gets a snapshot of a provider's data including credentials and alerts.",
+        allow_flagging="never"
+    )
+
+    demo = gr.TabbedInterface(
+        [iface_sync, iface_add_cred, iface_expiring, iface_snapshot],
+        ["Sync Provider", "Add/Update Credential", "List Expiring", "Provider Snapshot"]
+    )
+
+    # Launch with mcp_server=True to enable MCP endpoints
+    demo.launch(mcp_server=True)
+
+if __name__ == "__main__":
+    main()

src/cred_db_mcp_server/schemas.py

@@ -0,0 +1,29 @@
+from pydantic import BaseModel
+from typing import Optional, List
+
+# Provider Schema
+class Provider(BaseModel):
+    id: int
+    npi: str
+    name: str
+    # Add other fields as necessary based on CRED_API response
+
+class Credential(BaseModel):
+    id: Optional[int] = None
+    provider_id: int
+    type: str
+    issuer: str
+    number: str
+    expiry_date: str
+    # Add other fields as necessary
+
+class ExpiringCredential(BaseModel):
+    provider: Provider
+    credential: Credential
+    days_to_expiry: int
+    risk_score: float
+
+class ProviderSnapshot(BaseModel):
+    provider: Provider
+    credentials: List[Credential]
+    alerts: Optional[List[dict]] = None

src/cred_db_mcp_server/tools.py

@@ -0,0 +1,152 @@
+"""
+This module implements the tools exposed by the MCP server.
+Each function corresponds to an MCP tool.
+"""
+import httpx
+from typing import Optional, List, Dict, Any
+from .config import CRED_API_BASE_URL
+# from .schemas import Provider, Credential, ExpiringCredential, ProviderSnapshot
+
+# We can use simple types in signature for Gradio MCP compatibility
+# but use schemas for internal validation if needed.
+# Since the prompt asked for mapped errors, we'll wrap calls.
+
+def sync_provider_from_npi(npi: str) -> Dict[str, Any]:
+    """
+    Syncs a provider's data from the NPI registry.
+
+    Args:
+        npi (str): The NPI number of the provider.
+
+    Returns:
+        dict: The provider object returned by the API.
+    """
+    url = f"{CRED_API_BASE_URL}/providers/sync_from_npi"
+    try:
+        with httpx.Client() as client:
+            response = client.post(url, json={"npi": npi})
+            response.raise_for_status()
+            return response.json()
+    except httpx.HTTPStatusError as e:
+        # Map HTTP errors to MCP/User friendly errors
+        if e.response.status_code == 404:
+            return {"error": f"Provider with NPI {npi} not found or sync failed."}
+        return {"error": f"API Error: {e.response.text}"}
+    except Exception as e:
+        return {"error": f"Connection Error: {str(e)}"}
+
+def add_or_update_credential(
+    provider_id: int,
+    type: str,
+    issuer: str,
+    number: str,
+    expiry_date: str
+) -> Dict[str, Any]:
+    """
+    Adds or updates a credential for a provider.
+
+    Args:
+        provider_id (int): The internal ID of the provider.
+        type (str): The type of credential (e.g., 'Medical License').
+        issuer (str): The issuing body (e.g., 'State Board').
+        number (str): The credential number.
+        expiry_date (str): The expiry date in YYYY-MM-DD format.
+
+    Returns:
+        dict: The created or updated credential object.
+    """
+    url = f"{CRED_API_BASE_URL}/credentials/add_or_update"
+    payload = {
+        "provider_id": provider_id,
+        "type": type,
+        "issuer": issuer,
+        "number": number,
+        "expiry_date": expiry_date
+    }
+    try:
+        with httpx.Client() as client:
+            response = client.post(url, json=payload)
+            response.raise_for_status()
+            return response.json()
+    except httpx.HTTPStatusError as e:
+        return {"error": f"API Error: {e.response.text}"}
+    except Exception as e:
+        return {"error": f"Connection Error: {str(e)}"}
+
+def list_expiring_credentials(
+    window_days: int,
+    dept: Optional[str] = None,
+    location: Optional[str] = None
+) -> List[Dict[str, Any]]:
+    """
+    Lists credentials expiring within a certain number of days.
+
+    Args:
+        window_days (int): The number of days to check for expiry.
+        dept (str, optional): Filter by department.
+        location (str, optional): Filter by location.
+
+    Returns:
+        list: A list of objects containing provider, credential, days_to_expiry, and risk_score.
+    """
+    url = f"{CRED_API_BASE_URL}/credentials/expiring"
+    payload = {
+        "window_days": window_days,
+        "dept": dept,
+        "location": location
+    }
+    # Remove None values to avoid sending them if API doesn't expect them or treat them as valid
+    payload = {k: v for k, v in payload.items() if v is not None}
+
+    try:
+        with httpx.Client() as client:
+            response = client.post(url, json=payload)
+            response.raise_for_status()
+            return response.json()
+    except httpx.HTTPStatusError as e:
+        # In case of error, return a list with an error dict or raise
+        # For MCP, returning structured error info is usually better than crashing
+        # But for list return type, we might need to handle differently.
+        # Here we assume the tool call handles exceptions or checks for "error" key in result if it was a dict.
+        # Since return type is List, we can't easily return a dict error.
+        # We'll return an empty list and print error or raise.
+        # Let's raise ValueError which Gradio might catch and show.
+        raise ValueError(f"API Error: {e.response.text}")
+    except Exception as e:
+        raise ConnectionError(f"Connection Error: {str(e)}")
+
+def get_provider_snapshot(
+    provider_id: Optional[int] = None,
+    npi: Optional[str] = None
+) -> Dict[str, Any]:
+    """
+    Gets a snapshot of a provider's data including credentials and alerts.
+
+    Args:
+        provider_id (int, optional): The provider's internal ID.
+        npi (str, optional): The provider's NPI.
+
+    Returns:
+        dict: Object containing provider details, credentials, and alerts.
+    """
+    if provider_id is None and npi is None:
+        return {"error": "Must provide either provider_id or npi."}
+
+    url = f"{CRED_API_BASE_URL}/providers/snapshot"
+    payload = {}
+    if provider_id is not None:
+        payload["provider_id"] = provider_id
+    if npi is not None:
+        payload["npi"] = npi
+
+    try:
+        with httpx.Client() as client:
+            response = client.post(url, json=payload)
+            response.raise_for_status()
+            return response.json()
+    except httpx.HTTPStatusError as e:
+        if e.response.status_code == 404:
+            return {"error": "Provider not found."}
+        return {"error": f"API Error: {e.response.text}"}
+    except Exception as e:
+        return {"error": f"Connection Error: {str(e)}"}

tests/test_cred_db_mcp_server.py

@@ -0,0 +1,86 @@
+
+import pytest
+from unittest.mock import patch, MagicMock
+from cred_db_mcp_server.tools import (
+    sync_provider_from_npi,
+    add_or_update_credential,
+    list_expiring_credentials,
+    get_provider_snapshot
+)
+import httpx
+
+# Mock response data
+MOCK_PROVIDER = {"id": 1, "npi": "1234567890", "name": "Dr. Test"}
+MOCK_CREDENTIAL = {
+    "id": 10,
+    "provider_id": 1,
+    "type": "Medical License",
+    "issuer": "State Board",
+    "number": "MD12345",
+    "expiry_date": "2025-01-01"
+}
+MOCK_EXPIRING_LIST = [
+    {
+        "provider": MOCK_PROVIDER,
+        "credential": MOCK_CREDENTIAL,
+        "days_to_expiry": 30,
+        "risk_score": 0.5
+    }
+]
+MOCK_SNAPSHOT = {
+    "provider": MOCK_PROVIDER,
+    "credentials": [MOCK_CREDENTIAL],
+    "alerts": []
+}
+
+@patch("cred_db_mcp_server.tools.httpx.Client")
+def test_sync_provider_from_npi(mock_client_cls):
+    mock_client = MagicMock()
+    mock_client_cls.return_value.__enter__.return_value = mock_client
+    mock_client.post.return_value.status_code = 200
+    mock_client.post.return_value.json.return_value = MOCK_PROVIDER
+
+    result = sync_provider_from_npi("1234567890")
+    assert result == MOCK_PROVIDER
+    mock_client.post.assert_called_once()
+    assert "sync_from_npi" in mock_client.post.call_args[0][0]
+
+@patch("cred_db_mcp_server.tools.httpx.Client")
+def test_add_or_update_credential(mock_client_cls):
+    mock_client = MagicMock()
+    mock_client_cls.return_value.__enter__.return_value = mock_client
+    mock_client.post.return_value.status_code = 200
+    mock_client.post.return_value.json.return_value = MOCK_CREDENTIAL
+
+    result = add_or_update_credential(1, "Medical License", "State Board", "MD12345", "2025-01-01")
+    assert result == MOCK_CREDENTIAL
+    mock_client.post.assert_called_once()
+    assert "add_or_update" in mock_client.post.call_args[0][0]
+
+@patch("cred_db_mcp_server.tools.httpx.Client")
+def test_list_expiring_credentials(mock_client_cls):
+    mock_client = MagicMock()
+    mock_client_cls.return_value.__enter__.return_value = mock_client
+    mock_client.post.return_value.status_code = 200
+    mock_client.post.return_value.json.return_value = MOCK_EXPIRING_LIST
+
+    result = list_expiring_credentials(30)
+    assert result == MOCK_EXPIRING_LIST
+    mock_client.post.assert_called_once()
+    assert "expiring" in mock_client.post.call_args[0][0]
+
+@patch("cred_db_mcp_server.tools.httpx.Client")
+def test_get_provider_snapshot(mock_client_cls):
+    mock_client = MagicMock()
+    mock_client_cls.return_value.__enter__.return_value = mock_client
+    mock_client.post.return_value.status_code = 200
+    mock_client.post.return_value.json.return_value = MOCK_SNAPSHOT
+
+    result = get_provider_snapshot(provider_id=1)
+    assert result == MOCK_SNAPSHOT
+    mock_client.post.assert_called_once()
+    assert "snapshot" in mock_client.post.call_args[0][0]
+
+def test_get_provider_snapshot_no_args():
+    result = get_provider_snapshot()
+    assert "error" in result