Introduction

Hey there, Claude enthusiasts! If you've been diving into the Anthropic ecosystem, you've probably heard about MCP (Model Context Protocol) servers. These bad boys are game-changers for extending Claude's capabilities beyond its native smarts. Imagine Claude querying your live Postgres database or caching hot data in Redis—all in real-time, without you lifting a finger after setup.

In this advanced tutorial, we'll build custom MCP servers for Postgres and Redis integration. We'll cover everything from scratch: server setup, database tools, Docker deployment, and Claude-specific prompts for seamless tool-calling. Whether you're a dev building AI agents or a team automating workflows, this will solve real problems like dynamic data retrieval in your Claude-powered apps.

By the end, you'll have a production-ready MCP server running in Docker, ready to plug into Claude Code, the API, or agents. Let's dive in!

What Are MCP Servers?

MCP servers act as intermediaries between Claude and external resources. They implement the Model Context Protocol, a lightweight HTTP-based spec from Anthropic that lets Claude call custom tools via JSON schemas. Think of it like OpenAI's function calling but optimized for Claude's Opus, Sonnet, or Haiku models.

Key benefits:

• Real-time data access: No more static context windows—Claude fetches fresh DB data on-demand.
• Security: Servers validate and sanitize queries, keeping your DB safe.
• Scalability: Docker + cloud deployment for enterprise teams.
• Claude-native: Works out-of-the-box with Claude Code CLI and API SDK.

Claude parses your tool schemas, decides when to call them, and handles the JSON response loop automatically.

Prerequisites

Before we code, grab these:

• Python 3.10+ and pip
• Docker and Docker Compose
• Postgres 15+ (local or cloud like Supabase)
• Redis 7+ (local or Redis Cloud)
• Claude API key (from console.anthropic.com)
• Familiarity with FastAPI (we'll use it for the MCP server)

Install deps:

pip install fastapi uvicorn psycopg2-binary redis pydantic python-dotenv

Set up env vars in a .env file:

POSTGRES_URL=postgresql://user:pass@localhost:5432/mydb
REDIS_URL=redis://localhost:6379
CLAUDE_API_KEY=your_key_here

Building a Basic MCP Server

We'll use FastAPI for its async speed and auto-docs—perfect for MCP. Start with a minimal server.

Create mcp_server.py:

import os
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Any, Dict, List
import asyncpg
import redis.asyncio as redis
from dotenv import load_dotenv

load_dotenv()

app = FastAPI(title="Claude MCP Server")

# MCP endpoint: Claude POSTs tool calls here
@app.post("/mcp/call")
async def handle_tool_call(tool_call: Dict[str, Any]):
    name = tool_call.get("name")
    args = tool_call.get("arguments", {})
    
    if name == "query_postgres":
        return await query_postgres(args)
    elif name == "get_redis":
        return await get_redis(args)
    else:
        raise HTTPException(status_code=400, detail="Unknown tool")

async def query_postgres(args: Dict[str, Any]) -> Dict[str, Any]:
    query = args.get("query")
    conn = await asyncpg.connect(os.getenv("POSTGRES_URL"))
    try:
        rows = await conn.fetch(query)
        return {"result": [dict(row) for row in rows]}
    finally:
        await conn.close()

async def get_redis(args: Dict[str, Any]) -> Dict[str, Any]:
    key = args.get("key")
    r = redis.from_url(os.getenv("REDIS_URL"))
    try:
        value = await r.get(key)
        return {"value": value.decode() if value else None}
    finally:
        await r.aclose()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Run it:

uvicorn mcp_server:app --reload

Test at http://localhost:8000/docs.

Postgres Integration: Secure Query Tool

Postgres is king for structured data. Our query_postgres tool lets Claude run SQL—but safely.

Enhance security in query_postgres:

# Add to mcp_server.py
SAFE_QUERIES = ["SELECT", "SHOW"]  # Whitelist prefixes

async def query_postgres(args: Dict[str, Any]) -> Dict[str, Any]:
    query = args["query"].strip().upper()
    if not any(query.startswith(sq) for sq in SAFE_QUERIES):
        raise HTTPException(403, "Unsafe query")
    # ... rest as before

Claude tool schema (you'll define this in prompts):

{
  "name": "query_postgres",
  "description": "Execute a read-only SQL query on Postgres DB.",
  "input_schema": {
    "type": "object",
    "properties": {
      "query": {"type": "string", "description": "Safe SELECT query"}
    },
    "required": ["query"]
  }
}

Redis Integration: Real-Time Caching

Redis shines for speed. Use it for live metrics, sessions, or pub/sub.

Add a set tool too:

async def set_redis(args: Dict[str, Any]) -> Dict[str, Any]:
    key = args["key"]
    value = args["value"]
    ttl = args.get("ttl", 3600)
    r = redis.from_url(os.getenv("REDIS_URL"))
    await r.set(key, value, ex=ttl)
    await r.aclose()
    return {"status": "set"}

# Update handle_tool_call
if name == "set_redis":
    return await set_redis(args)

Schema:

{
  "name": "get_redis",
  "description": "Get value from Redis key.",
  "input_schema": {
    "type": "object",
    "properties": {
      "key": {"type": "string"}
    }
  }
}
{
  "name": "set_redis",
  "description": "Set Redis key-value with optional TTL.",
  "input_schema": {
    "type": "object",
    "properties": {
      "key": {"type": "string"},
      "value": {"type": "string"},
      "ttl": {"type": "integer"}
    },
    "required": ["key", "value"]
  }
}

Dockerizing for Easy Deployment

Containerize for portability. Create 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", "8000"]

requirements.txt:

fastapi==0.104.1
uvicorn[standard]==0.24.0
asyncpg==0.29.0
redis==5.0.1
pydantic==2.5.0
python-dotenv==1.0.0

docker-compose.yml for DBs:

version: '3.8'
services:
  mcp-server:
    build: .
    ports:
      - "8000:8000"
    env_file: .env
    depends_on:
      - postgres
      - redis
  postgres:
    image: postgres:15
    environment:
      POSTGRES_DB: mydb
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
    ports:
      - "5432:5432"
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

Spin up:

docker-compose up -d

Now your MCP server runs at localhost:8000, backed by real DBs.

Integrating with Claude: Tool-Calling Prompts

Time to wire it into Claude! Use Claude Code CLI or API.

First, Claude Code setup:

claude-code init myproject
cd myproject

In your prompt or claude.md:

You have access to these MCP tools via http://host.docker.internal:8000/mcp/call

<tools>
{"name": "query_postgres", "description": "...", "input_schema": {...}}
{"name": "get_redis", ...}
{"name": "set_redis", ...}
</tools>

Task: Analyze sales data from DB and cache summary in Redis.

API example (Python SDK):

from anthropic import Anthropic
client = Anthropic(api_key=os.getenv("CLAUDE_API_KEY"))

tools = [  # Paste schemas here
    {"name": "query_postgres", ...},
]

message = client.messages.create(
    model="claude-3-5-sonnet-20240620",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "What's our top product? Cache it."}],
    tool_choice="auto",
)

# Handle tool calls
for tool in message.stop_reason == "tool_use":
    response = requests.post("http://localhost:8000/mcp/call", json=tool)
    # Feed back to Claude

Claude will output tool calls like:

{"name": "query_postgres", "arguments": {"query": "SELECT product, SUM(sales) FROM orders GROUP BY product ORDER BY sum DESC LIMIT 1;"}}

Advanced Prompts for Tool-Chaining

Pro tip: Chain tools for complex workflows.

Prompt:

Step 1: Query Postgres for user orders: SELECT * FROM orders WHERE user_id = {id}
Step 2: Compute total: Sum the amounts.
Step 3: Cache in Redis as 'user:{id}:total' with TTL 300.
Step 4: Report back.
Use tools only when needed. Be precise with SQL.

This leverages Claude's reasoning + real-time data.

Real-World Example: Inventory Agent

Let's build an agent for e-commerce inventory.

1. Claude queries Postgres: SELECT stock FROM products WHERE id = ?
2. If low, checks Redis for supplier ETA: get_redis('supplier:eta')
3. Updates cache: set_redis('product:{id}:alert', 'low_stock')

Prompt snippet:

You are InventoryBot. Monitor stock levels.
User: Check product 123.

Claude chains: query -> check threshold -> alert/cache.

Sample DB setup (psql):

CREATE TABLE products (id SERIAL PRIMARY KEY, name TEXT, stock INT);
INSERT INTO products (name, stock) VALUES ('Widget', 5), ('Gadget', 50);

Test: Claude pulls live stock, caches alerts—perfect for Slack/Zapier integrations.

Deployment and Scaling

For prod:

• Deploy to Fly.io, Render, or AWS ECS with Docker.
• Use HTTPS + API keys for auth.
• Scale with Redis Cluster/Postgres replicas.
• Monitor with Prometheus.

Add auth to MCP:

@app.post("/mcp/call")
async def handle_tool_call(tool_call: Dict, api_key: str = Header(None)):
    if api_key != os.getenv("MCP_API_KEY"):
        raise HTTPException(401)

Pass x-api-key in Claude's tool calls.

Best Practices

• Sanitize inputs: Always validate SQL/args.
• Rate limiting: Use FastAPI middleware.
• Error handling: Return Claude-friendly JSON.
• Model choice: Sonnet for reasoning, Haiku for speed.
• Context management: Cache frequent queries in Redis.
• Testing: Mock DBs in CI.

Common pitfalls: Vague tool descriptions lead to bad args—be explicit!

Wrapping Up

Boom! You've got a battle-tested MCP server bridging Claude to Postgres/Redis. Deploy it, hook into your agents, and watch productivity soar. Whether for HR analytics, sales dashboards, or engineering workflows, this setup scales.

Questions? Drop 'em in the comments. Star us on GitHub, and check claudedirectory.com for more Claude tools.

Happy building! 🚀

(Word count: ~1450)