Model Context Protocol

# server.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, Dict

app = FastAPI(title="customer-db-mcp", version="0.1.0")


# Request schema: what the MCP client (the AI agent) will send
class GetCustomerRequest(BaseModel):
    customer_id: str


# Response schema: what the MCP server returns
class Customer(BaseModel):
    customer_id: str
    name: str
    email: Optional[str] = None
    status: str  # e.g., 'shipped', 'processing', 'closed' (use values appropriate for your domain)


# Fake in-memory database (replace with real DB in production)
customers: Dict[str, Customer] = {
    "1234": Customer(customer_id="1234", name="Alice Johnson", email="alice@example.com", status="shipped"),
    "2345": Customer(customer_id="2345", name="Bob Smith", email="bob@example.com", status="processing"),
}


@app.post("/mcp/get_customer", response_model=Customer)
async def get_customer(req: GetCustomerRequest):
    """
    MCP function: returns customer information by customer_id.
    The agent can call this endpoint to retrieve customer state.
    """
    cust = customers.get(req.customer_id)
    if not cust:
        raise HTTPException(status_code=404, detail="customer not found")
    return cust

uvicorn server:app --host 0.0.0.0 --port 8000 --reload

curl -X POST "http://localhost:8000/mcp/get_customer" \
  -H "Content-Type: application/json" \
  -d '{"customer_id": "1234"}'

{
  "customer_id": "1234",
  "name": "Alice Johnson",
  "email": "alice@example.com",
  "status": "shipped"
}