Custom MCP Servers for Claude: Integrate Private Databases and APIs Seamlessly

# Unlock Claude's Power with Custom MCP Servers Hey there, Claude enthusiasts! If you've ever felt frustrated because Claude can't peek into your company's private databases or hit those internal APIs directly, you're not alone. That's where **Model Context Protocol (MCP) servers** come in – lightweight, secure bridges that let Claude fetch context from your private data sources on demand. In this guide, we'll build one from scratch using Python and FastAPI, connect it to a Postgres DB and a mock private API, deploy it, and wire it up to Claude. By the end, you'll have a production-ready setup boosting your agents' accuracy. Whether you're a dev automating sales workflows or a team evaluating Claude for enterprise, custom MCP servers solve real problems like data silos and compliance. Let's dive in! ## Why Bother with Custom MCP Servers? Before we code, here's the "why" in a quick list: - **Privacy First**: Keep sensitive data off Anthropic's servers – Claude requests it just-in-time. - **Real-Time Accuracy**: No more hallucinated facts; pull fresh data from your DB or API. - **Scalable Agents**: Perfect for AI agents in HR (employee records), Marketing (campaign metrics), or Engineering (bug trackers). - **Claude-Native**: Leverages Claude's tool-calling via the Anthropic SDK. - **Easy Deployment**: Host on Vercel, Fly.io, or your VPC in minutes. MCP follows a simple protocol: Claude sends a JSON query to your `/context` endpoint, you process it securely, and respond with structured data. Boom – context injected! ## Prerequisites: Gear Up in 5 Minutes Grab these: - Python 3.10+ - [FastAPI](https://fastapi.tiangolo.com/) and [Uvicorn](https://www.uvicorn.org/) - [SQLAlchemy](https://www.sqlalchemy.org/) for DB (or your ORM) - [Anthropic SDK](https://docs.anthropic.com/claude/reference/client-sdks) - Docker for deployment - A Postgres DB (local or cloud like Supabase) - ngrok for local testing Install deps: ```bash pip install fastapi uvicorn sqlalchemy psycopg2-binary anthropic pydantic python-dotenv ``` ## Step 1: Scaffold Your MCP Server Start with a basic FastAPI app. Create `mcp_server.py`: ```python import os from fastapi import FastAPI, HTTPException, Depends from fastapi.security import APIKeyHeader from pydantic import BaseModel from typing import Dict, Any import json from anthropic import Anthropic # We'll use this later for testing app = FastAPI(title="Custom MCP Server for Claude") # Security: API key header api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False) async def get_api_key(api_key: str = Depends(api_key_header)): if api_key != os.getenv("MCP_API_KEY"): raise HTTPException(status_code=401, detail="Invalid API key") return api_key class MCPQuery(BaseModel): query: str params: Dict[str, Any] = {} class MCPResponse(BaseModel): context: Dict[str, Any] sources: list[str] = [] @app.post("/context", response_model=MCPResponse) async def get_context(query: MCPQuery, api_key: str = Depends(get_api_key)): # Placeholder logic – we'll flesh this out return MCPResponse(context={"result": f"Processed: {query.query}"}, sources=["placeholder"]) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000) ``` Run it: `uvicorn mcp_server:app --reload`. Hit `http://localhost:8000/docs` for Swagger UI. Test with curl: ```bash curl -X POST "http://localhost:8000/context" \ -H "Content-Type: application/json" \ -H "X-API-Key: your-secret-key" \ -d '{"query": "fetch users", "params": {"limit": 5}}' ``` Set `MCP_API_KEY=your-secret-key` in `.env`. ## Step 2: Integrate a Private Database (Postgres Example) Let's connect to a Postgres DB for employee data – think HR playbook! First, add SQLAlchemy: ```python from sqlalchemy import create_engine, text from sqlalchemy.orm import sessionmaker # .env: DATABASE_URL=postgresql://user:pass@localhost/hr_db engine = create_engine(os.getenv("DATABASE_URL")) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) @app.post("/context") async def get_context(query: MCPQuery, api_key: str = Depends(get_api_key), db: SessionLocal = Depends(get_db)): if "employee" in query.query.lower(): stmt = text("SELECT id, name, department FROM employees WHERE department = :dept LIMIT :limit") result = db.execute(stmt, {"dept": query.params.get("dept", "Engineering"), "limit": query.params.get("limit", 10)}).fetchall() context = [dict(row._mapping) for row in result] return MCPResponse(context={"employees": context}, sources=["hr_db"]) raise HTTPException(status_code=400, detail="Unsupported query") # Dependency def get_db(): db = SessionLocal() try: yield db finally: db.close() ``` Pro tip: Use RLS (Row-Level Security) in Supabase for compliance. ## Step 3: Hook Up a Private API Now, proxy a private CRM API (e.g., internal Salesforce clone). ```python import httpx CRM_BASE_URL = os.getenv("CRM_URL", "https://internal-crm.example.com") @app.post("/context") async def get_context(query: MCPQuery, api_key: str = Depends(get_api_key), db=Depends(get_db)): if "leads" in query.query.lower(): async with httpx.AsyncClient() as client: resp = await client.get(f"{CRM_BASE_URL}/leads", params=query.params, headers={"Authorization": f"Bearer {os.getenv('CRM_TOKEN')}"}) if resp.status_code == 200: return MCPResponse(context=resp.json(), sources=["crm_api"]) # ... handle DB case too ``` This keeps tokens server-side – secure! ## Step 4: Advanced: Semantic Query Routing Make it smart. Route queries dynamically: ```python QUERY_ROUTERS = { "employee": handle_employee_query, "leads": handle_leads_query, # Add more } async def get_context(...): for key, handler in QUERY_ROUTERS.items(): if key in query.query.lower(): return await handler(query, db) raise HTTPException(status_code=400, detail="No handler for query") ``` Bonus: Cache with Redis for speed. ## Step 5: Secure It Like Enterprise - **API Keys**: Rotate with Vault. - **Rate Limiting**: Use `slowapi`. - **HTTPS Only**: Enforce in prod. - **CORS**: Restrict to your Claude app domain. - **Logging**: Structured with `structlog` for audits. Add rate limit: ```python from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(429, _rate_limit_exceeded_handler) @app.post("/context") @limiter.limit("10/minute") async def get_context(...): ... ``` ## Step 6: Deploy to Production Dockerize it: `Dockerfile`: ```dockerfile FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD ["uvicorn", "mcp_server:app", "--host", "0.0.0.0", "--port", "80"] ``` Deploy to Railway: 1. `railway login` 2. `railway new` 3. Push secrets: `railway secrets set MCP_API_KEY=...` URL: `https://your-mcp.railway.app` ## Step 7: Integrate with Claude API Now, teach Claude to use it! Define the tool in your agent code: ```python from anthropic import Anthropic client = Anthropic(api_key="your-anthropic-key") tools = [ { "name": "mcp_query", "description": "Query private data via MCP server", "input_schema": { "type": "object", "properties": { "query": {"type": "string"}, "params": {"type": "object"} } } } ] def mcp_tool_call(tool_call): # Call your MCP server resp = httpx.post("https://your-mcp.railway.app/context", json=tool_call["input"], headers={"X-API-Key": "your-secret-key"}) return resp.json()["context"] message = client.messages.create( model="claude-3-5-sonnet-20240620", max_tokens=1024, tools=tools, messages=[{"role": "user", "content": "Get top 5 engineering employees and recent leads."}], tool_choice="auto" ) # Handle tool use loop here print(message.content) ``` Claude will call `mcp_query`, you fetch data, append to context, and iterate. ## Real-World Examples - **Sales Playbook**: Query CRM leads by region. - **Engineering**: Fetch Jira tickets via private API. - **Legal**: Pull contract DB with semantic search. ## Best Practices & Gotchas - **Query Validation**: Sanitize to prevent SQLi. - **Error Handling**: Graceful fallbacks for Claude. - **Cost Optimization**: Batch queries. - **Monitoring**: Prometheus + Grafana. - **Scale**: Async handlers for high TPS. Common pitfall: Forgetting to handle tool_choice="auto" in loops. ## Wrapping Up Congrats – you've built a custom MCP server! This setup powers robust Claude agents accessing private data without compromises. Experiment with your stack (swap Postgres for Mongo, FastAPI for Express.js). Share your builds in comments or on claudedirectory.com forums. Next reads: [Claude Code CLI Deep Dive](/claude-code-cli) | [Building Agents with MCP](/agents-mcp). Word count: ~1450. Questions? Hit reply!