Implement cred_db_mcp server with FastAPI and SQLite

- Set up project structure with uv/pyproject.toml
- Implement SQLAlchemy 2.0 models for Providers and Credentials
- Implement MCP tools: sync_provider, add_credential, list_expiring, get_snapshot
- Expose tools via FastAPI endpoints
- Add comprehensive tests for DB logic

.gitignore

@@ -0,0 +1,5 @@
+__pycache__/
+*.pyc
+*.db
+.env
+.pytest_cache/

README.md

@@ -0,0 +1,48 @@
+
+# Usage of cred_db_mcp
+
+This server provides direct database access and logic for **CredentialWatch** via MCP-compliant tools exposed as HTTP endpoints.
+
+## Agents Usage
+
+Agents can use this server to:
+1.  **Onboard new providers**: Call `sync_provider_from_npi` to fetch details from the NPI registry and create a local record.
+2.  **Manage Credentials**: Use `add_or_update_credential` to keep license data up to date.
+3.  **Monitor Compliance**: Use `list_expiring_credentials` to proactively find providers who need to renew licenses.
+4.  **Context Retrieval**: Use `get_provider_snapshot` to get all known data about a provider before answering user questions.
+
+## Running the Server
+
+```bash
+# Install dependencies
+pip install -e .
+
+# Run
+uvicorn src.cred_db_mcp.main:app --reload
+```
+
+## Example Tool Call
+
+**Tool:** `list_expiring_credentials`
+**Endpoint:** `POST /mcp/tools/list_expiring_credentials`
+**Headers:** `Content-Type: application/json`
+
+**Body:**
+```json
+{
+  "window_days": 90,
+  "dept": "Cardiology"
+}
+```
+
+**Response:**
+```json
+[
+  {
+    "provider": { "full_name": "Dr. Alice Smith", ... },
+    "credential": { "type": "state_license", "expiry_date": "2023-12-01", ... },
+    "days_to_expiry": 25,
+    "risk_score": 3
+  }
+]
+```

pyproject.toml

@@ -0,0 +1,27 @@
+[project]
+name = "cred_db_mcp"
+version = "0.1.0"
+description = "Credential Database MCP Server"
+requires-python = ">=3.11"
+dependencies = [
+    "fastapi>=0.100.0",
+    "uvicorn>=0.20.0",
+    "sqlalchemy>=2.0.0",
+    "pydantic>=2.0.0",
+    "httpx>=0.24.0",
+    "python-dotenv>=1.0.0"
+]
+
+[project.optional-dependencies]
+test = [
+    "pytest>=7.0.0",
+    "pytest-asyncio>=0.21.0"
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.pytest.ini_options]
+pythonpath = "src"
+testpaths = ["tests"]

src/cred_db_mcp/db.py

@@ -0,0 +1,33 @@
+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

@@ -0,0 +1,98 @@
+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

@@ -0,0 +1,185 @@
+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

@@ -0,0 +1,62 @@
+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

@@ -0,0 +1,40 @@
+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

@@ -0,0 +1,79 @@
+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

tests/mock_npi_mcp.py

@@ -0,0 +1,22 @@
+from fastapi import FastAPI
+from pydantic import BaseModel
+
+app = FastAPI()
+
+class NpiRequest(BaseModel):
+    npi: str
+
+@app.post("/mcp/tools/get_provider_by_npi")
+def get_provider_by_npi(req: NpiRequest):
+    if req.npi == "1234567890":
+        return {
+            "npi": "1234567890",
+            "first_name": "John",
+            "last_name": "Doe",
+            "taxonomy_desc": "Cardiology",
+            "practice_address": {
+                "city": "New York",
+                "state": "NY"
+            }
+        }
+    return None

tests/test_cred_db_mcp.py

@@ -0,0 +1,152 @@
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
+from datetime import date, timedelta
+import os
+
+from src.cred_db_mcp.main import app, get_db
+from src.cred_db_mcp.db import Base
+from src.cred_db_mcp.models import Provider, Credential
+
+# Setup in-memory DB for tests
+# Use StaticPool to share the in-memory DB across multiple sessions/connections
+SQLALCHEMY_DATABASE_URL = "sqlite:///:memory:"
+
+engine = create_engine(
+    SQLALCHEMY_DATABASE_URL,
+    connect_args={"check_same_thread": False},
+    poolclass=StaticPool
+)
+TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+def override_get_db():
+    try:
+        db = TestingSessionLocal()
+        yield db
+    finally:
+        db.close()
+
+app.dependency_overrides[get_db] = override_get_db
+
+@pytest.fixture(scope="module")
+def test_client():
+    # Create tables once for the module
+    Base.metadata.create_all(bind=engine)
+    client = TestClient(app)
+    yield client
+    Base.metadata.drop_all(bind=engine)
+
+@pytest.fixture(autouse=True)
+def clean_tables():
+    # Optional: Clear data between tests if needed, but for now just appending is fine
+    # as long as IDs/NPIs don't clash or we don't care about accumulation.
+    # To be safer, we can delete all data.
+    with engine.connect() as conn:
+        conn.execute(Credential.__table__.delete())
+        conn.execute(Provider.__table__.delete())
+        conn.commit()
+
+@pytest.fixture
+def db_session():
+    db = TestingSessionLocal()
+    yield db
+    db.close()
+
+def test_read_root(test_client):
+    response = test_client.get("/")
+    assert response.status_code == 200
+    assert response.json()["service"] == "cred_db_mcp"
+
+def test_add_and_snapshot_provider(test_client, db_session):
+    prov = Provider(
+        npi="999999", full_name="Test Doc", primary_specialty="General", is_active=True
+    )
+    db_session.add(prov)
+    db_session.commit()
+
+    response = test_client.post(
+        "/mcp/tools/get_provider_snapshot",
+        json={"npi": "999999"}
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["provider"]["full_name"] == "Test Doc"
+    assert len(data["credentials"]) == 0
+
+def test_add_credential(test_client, db_session):
+    # Seed provider
+    prov = Provider(
+        npi="888888", full_name="Credential Doc", primary_specialty="Surgery", is_active=True
+    )
+    db_session.add(prov)
+    db_session.commit()
+    db_session.refresh(prov)
+
+    # Add Credential via tool
+    expiry = (date.today() + timedelta(days=100)).strftime("%Y-%m-%d")
+    response = test_client.post(
+        "/mcp/tools/add_or_update_credential",
+        json={
+            "provider_id": prov.id,
+            "type": "board_cert",
+            "issuer": "ABMS",
+            "number": "XYZ123",
+            "expiry_date": expiry
+        }
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["number"] == "XYZ123"
+    assert data["status"] == "active"
+
+def test_list_expiring(test_client, db_session):
+    # Seed provider
+    prov = Provider(
+        npi="777777", full_name="Expiring Doc", dept="ER", location="NYC", is_active=True
+    )
+    db_session.add(prov)
+    db_session.commit()
+    db_session.refresh(prov)
+
+    # Add expiring credential (in 10 days)
+    # expiry date must be a date object for the model
+    expiry_1 = date.today() + timedelta(days=10)
+    cred = Credential(
+        provider_id=prov.id,
+        type="license",
+        issuer="State",
+        number="L1",
+        expiry_date=expiry_1,
+        status="active"
+    )
+    db_session.add(cred)
+
+    # Add non-expiring credential (in 100 days)
+    expiry_2 = date.today() + timedelta(days=100)
+    cred2 = Credential(
+        provider_id=prov.id,
+        type="license",
+        issuer="State",
+        number="L2",
+        expiry_date=expiry_2,
+        status="active"
+    )
+    db_session.add(cred2)
+    db_session.commit()
+
+    # Test tool
+    response = test_client.post(
+        "/mcp/tools/list_expiring_credentials",
+        json={
+            "window_days": 30,
+            "dept": "ER"
+        }
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert len(data) == 1
+    assert data[0]["credential"]["number"] == "L1"
+    assert data[0]["days_to_expiry"] == 10
+    assert data[0]["risk_score"] == 3  # < 30 days