Struggling to give Claude access to your private databases or APIs? Build custom MCP servers to integrate them seamlessly and supercharge your AI agents with real-time internal data.

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 and Uvicorn
• SQLAlchemy for DB (or your ORM)
• Anthropic SDK
• Docker for deployment
• A Postgres DB (local or cloud like Supabase)
• ngrok for local testing

Install deps:

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:

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:

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:

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).

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:

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:

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:

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:

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 | Building Agents with MCP.

Word count: ~1450. Questions? Hit reply!