Your App
    ├─ Custom code for PostgreSQL (50 lines)
    ├─ Custom code for Qdrant (50 lines)
    ├─ Custom code for Pinecone (50 lines)
    ├─ Custom code for REST API (50 lines)
    └─ Custom code for Redis (50 lines)

Total: 250 lines of custom integration code

Add a 6th tool? Write 50 more lines.
Change tools? Rewrite everything.
Use with LangGraph? Add 50 more lines per framework.

Result: 350+ lines of custom code just to connect tools.

Your App
    ├─ PostgreSQL MCP (100 lines, write once)
    ├─ Qdrant MCP (100 lines, write once)
    ├─ Pinecone MCP (100 lines, write once)
    ├─ REST API MCP (100 lines, write once)
    └─ Redis MCP (100 lines, write once)

Add a 6th tool? Write 100 lines for new MCP.
Change tools? Just swap MCPs, no app code changes.
Use with LangGraph? Works automatically.
Use with AutoGen? Works automatically.

Result: Write once per tool, works everywhere.

User: "Find customers named John"
    ↓
Claude (Client): "I need to call search_database tool"
    ↓
MCP Client → MCP Server (JSON-RPC request)
    {
        "jsonrpc": "2.0",
        "method": "tools/call",
        "params": {
            "name": "search_database",
            "arguments": {"name": "John"}
        }
    }
    ↓
MCP Server: Connects to PostgreSQL
    ↓
PostgreSQL: Returns [John Doe, john@example.com, ...]
    ↓
MCP Server → MCP Client (JSON response)
    {
        "result": {
            "content": [{
                "type": "text",
                "text": "Found: John Doe, john@example.com"
            }]
        }
    }
    ↓
Claude: "I got the results. Here's the answer..."

Tool: query_database
Description: "Search the database for customers"
Parameters:
  - name: "John" (required)
  - limit: 10 (optional)

Result: List of customers matching "John"

Claude: "What can you do?"

MCP Server responds:
{
    "tools": [
        {"name": "query_database", "description": "..."},
        {"name": "insert_record", "description": "..."},
        {"name": "delete_record", "description": "..."}
    ],
    "resources": [
        {"uri": "file:///data/customers.csv", "description": "..."}
    ]
}

Claude: "Great! I can call these 3 tools and access this file."

{
    "type": "object",
    "properties": {
        "name": {
            "type": "string",
            "description": "Customer name to search for"
        },
        "limit": {
            "type": "integer",
            "description": "Max results (default: 10)"
        }
    },
    "required": ["name"]
}

Valid call:   search_database(name="John", limit=5) ✓
Invalid call: search_database()                     ✗ (missing required 'name')

Success:
{ "content": [{"type":"text","text":"Found 3 customers"}], "isError": false }

Error:
{ "content": [{"type":"text","text":"Database connection failed"}], "isError": true }

pip install mcp

npm install @modelcontextprotocol/sdk

my-mcp-project/
├── my_mcp_server.py        (your MCP server)
├── requirements.txt         (dependencies)
├── config.json             (Claude Desktop config)
└── README.md               (documentation)

{
  "mcpServers": {
    "my-database": {
      "command": "python",
      "args": ["/path/to/my_mcp_server.py"]
    }
  }
}

import asyncio
import json
from mcp.server import Server
from mcp.types import Tool, TextContent, ToolResult

server = Server("my-server")

@server.tool()
async def my_first_tool(param1: str, param2: int = 10) -> ToolResult:
    """This is what your tool does."""
    try:
        result = f"Processed: {param1} with {param2}"
        return ToolResult(
            content=[TextContent(text=result)],
            is_error=False
        )
    except Exception as e:
        return ToolResult(
            content=[TextContent(text=f"Error: {str(e)}")],
            is_error=True
        )

server.register_tool(
    Tool(
        name="my_first_tool",
        description="Describe what this tool does",
        inputSchema={
            "type": "object",
            "properties": {
                "param1": {"type": "string", "description": "What is param1?"},
                "param2": {"type": "integer", "description": "Optional (default: 10)"}
            },
            "required": ["param1"]
        }
    )
)

if __name__ == "__main__":
    async def main():
        async with server:
            print("MCP Server is running...")
            await asyncio.sleep(float('inf'))
    asyncio.run(main())

import asyncio
import psycopg2
from psycopg2.extras import RealDictCursor
from mcp.server import Server
from mcp.types import Tool, TextContent, ToolResult

server = Server("database-mcp")

def get_db_connection():
    return psycopg2.connect(
        host="localhost", database="mydb",
        user="postgres", password="password"
    )

@server.tool()
async def query_database(sql: str, limit: int = 10) -> ToolResult:
    """Execute a SELECT query on the database"""
    try:
        conn = get_db_connection()
        cursor = conn.cursor(cursor_factory=RealDictCursor)
        cursor.execute(f"{sql} LIMIT {limit}")
        rows = cursor.fetchall()
        cursor.close(); conn.close()

        if not rows:
            return ToolResult(content=[TextContent(text="No results found")])

        result_text = "\n".join([str(dict(row)) for row in rows])
        return ToolResult(content=[TextContent(text=result_text)])
    except Exception as e:
        return ToolResult(
            content=[TextContent(text=f"Query failed: {str(e)}")],
            is_error=True
        )

import asyncio, httpx
from mcp.server import Server
from mcp.types import Tool, TextContent, ToolResult

server = Server("api-mcp")

@server.tool()
async def call_external_api(endpoint: str, method: str = "GET",
                            data: dict = None) -> ToolResult:
    """Call an external REST API"""
    try:
        async with httpx.AsyncClient() as client:
            url = f"https://api.example.com/{endpoint}"
            if method == "GET":
                response = await client.get(url)
            elif method == "POST":
                response = await client.post(url, json=data)
            else:
                return ToolResult(
                    content=[TextContent(text=f"Unsupported method: {method}")],
                    is_error=True
                )
            return ToolResult(content=[TextContent(text=str(response.json()))])
    except Exception as e:
        return ToolResult(
            content=[TextContent(text=f"API call failed: {str(e)}")],
            is_error=True
        )

import asyncio, random
from mcp.server import Server
from mcp.types import Tool, TextContent, ToolResult

server = Server("temperature-sensor")

def read_temperature():
    return round(random.uniform(15, 35), 1)

def read_humidity():
    return round(random.uniform(30, 80), 1)

@server.tool()
async def get_temperature() -> ToolResult:
    """Read current temperature from sensor"""
    try:
        temp = read_temperature()
        return ToolResult(content=[
            TextContent(text=f"Current temperature: {temp}°C")
        ])
    except Exception as e:
        return ToolResult(
            content=[TextContent(text=f"Error: {str(e)}")],
            is_error=True
        )

@server.tool()
async def check_temperature_alert(threshold: float = 30.0) -> ToolResult:
    """Check if temperature exceeds a threshold"""
    try:
        temp = read_temperature()
        humidity = read_humidity()
        if temp > threshold:
            msg = f"ALERT! {temp}°C exceeds {threshold}°C"
        else:
            msg = f"OK. {temp}°C is below {threshold}°C"
        msg += f"\nHumidity: {humidity}%"
        return ToolResult(content=[TextContent(text=msg)])
    except Exception as e:
        return ToolResult(
            content=[TextContent(text=f"Error: {str(e)}")],
            is_error=True
        )

server.register_tool(Tool(
    name="get_temperature",
    description="Read current temperature from sensor",
    inputSchema={"type": "object", "properties": {}, "required": []}
))
server.register_tool(Tool(
    name="check_temperature_alert",
    description="Check if temperature exceeds threshold",
    inputSchema={
        "type": "object",
        "properties": {
            "threshold": {"type": "number", "description": "Celsius threshold"}
        },
        "required": []
    }
))

if __name__ == "__main__":
    async def main():
        async with server:
            print("Temperature Sensor MCP running...")
            await asyncio.sleep(float('inf'))
    asyncio.run(main())

@server.tool()
async def my_tool(param1: str, param2: int = 10) -> ToolResult:
    """One-line description of what this tool does."""
    try:
        result = do_something(param1, param2)
        return ToolResult(
            content=[TextContent(text=str(result))],
            is_error=False
        )
    except Exception as e:
        return ToolResult(
            content=[TextContent(text=f"Error: {str(e)}")],
            is_error=True
        )

{
  "mcpServers": {
    "my-database": {
      "command": "python",
      "args": ["/full/path/to/my_mcp_server.py"]
    },
    "my-api": {
      "command": "python",
      "args": ["/full/path/to/api_mcp.py"]
    }
  }
}

from anthropic import Anthropic

client = Anthropic()
conversation_history = []

def chat_with_mcp(user_message):
    conversation_history.append({"role": "user", "content": user_message})
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2048,
        messages=conversation_history
    )
    assistant_message = response.content[0].text
    conversation_history.append({"role": "assistant", "content": assistant_message})
    return assistant_message

result = chat_with_mcp("Find customers in the database")

from langgraph.graph import StateGraph, START, END
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
import subprocess, json

@tool
def database_tool(query: str) -> str:
    """Search the database using MCP"""
    request = {
        "jsonrpc": "2.0",
        "method": "tools/call",
        "params": {
            "name": "query_database",
            "arguments": {"query": query}
        }
    }
    result = subprocess.run(
        ["python", "my_mcp_server.py"],
        input=json.dumps(request),
        capture_output=True, text=True
    )
    return result.stdout

llm = ChatAnthropic(model="claude-sonnet-4-20250514")
tools = [database_tool]
llm_with_tools = llm.bind_tools(tools)

def agent_step(state):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": state["messages"] + [response]}

graph = StateGraph(AgentState)
graph.add_node("agent", agent_step)
graph.add_edge(START, "agent")
graph.add_edge("agent", END)
agent = graph.compile()

result = agent.invoke({
    "messages": [HumanMessage(content="Find customers named John")]
})

from autogen import AssistantAgent, UserProxyAgent
import subprocess, json

def query_database(query: str) -> str:
    """Call MCP database tool"""
    request = {
        "jsonrpc": "2.0",
        "method": "tools/call",
        "params": {"name": "query_database", "arguments": {"query": query}}
    }
    result = subprocess.run(
        ["python", "my_mcp_server.py"],
        input=json.dumps(request),
        capture_output=True, text=True
    )
    return result.stdout

assistant = AssistantAgent(
    name="Database Assistant",
    system_message="You are a helpful assistant with database access.",
    llm_config={"config_list": [{"model": "gpt-4", "api_key": "YOUR_KEY"}]}
)
user_proxy = UserProxyAgent(name="User", human_input_mode="ALWAYS")
assistant.register_function(function_map={"query_database": query_database})
user_proxy.initiate_chat(assistant, message="Find all customers named John")

{
  "mcpServers": {
    "database": {"command": "python", "args": ["/path/to/database_mcp.py"]},
    "vectordb": {"command": "python", "args": ["/path/to/vector_mcp.py"]},
    "api":      {"command": "python", "args": ["/path/to/api_mcp.py"]},
    "files":    {"command": "python", "args": ["/path/to/files_mcp.py"]}
  }
}

db_query()       db_insert()       db_update()
vector_search()  vector_embed()
api_fetch()      api_post()
file_read()      file_write()

Prefixes help Claude know which MCP to call.

{ "mcpServers": { "my-mcp": { "command": "python", "args": ["/path/to/server.py"] } } }

FROM python:3.11-slim
WORKDIR /app
COPY my_mcp_server.py .
COPY requirements.txt .
RUN pip install -r requirements.txt
CMD ["python", "my_mcp_server.py"]

docker build -t my-mcp:latest .
docker run -p 8000:8000 my-mcp:latest

[Unit]
Description=My MCP Server
After=network.target

[Service]
Type=simple
User=mcpuser
WorkingDirectory=/opt/my-mcp
ExecStart=/usr/bin/python3 /opt/my-mcp/my_mcp_server.py
Restart=on-failure
RestartSec=5s

[Install]
WantedBy=multi-user.target

import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@server.tool()
async def health_check() -> ToolResult:
    """Check if MCP server is healthy"""
    try:
        conn = get_db_connection()
        conn.close()
        return ToolResult(content=[TextContent(text="Server healthy")])
    except Exception as e:
        return ToolResult(
            content=[TextContent(text=f"Server error: {str(e)}")],
            is_error=True
        )

@server.tool()
async def my_tool(param: str) -> ToolResult:
    try:
        if not param:
            return ToolResult(
                content=[TextContent(text="Parameter cannot be empty")],
                is_error=True)
        result = do_work(param)
        return ToolResult(content=[TextContent(text=result)])
    except ValueError as e:
        logger.error(f"Invalid value: {str(e)}")
        return ToolResult(
            content=[TextContent(text=f"Invalid input: {str(e)}")],
            is_error=True)
    except Exception as e:
        logger.error(f"Unexpected error: {str(e)}", exc_info=True)
        return ToolResult(
            content=[TextContent(text="An error occurred")],
            is_error=True)

import pytest

@pytest.mark.asyncio
async def test_query_database_success():
    result = await query_database("SELECT * FROM customers")
    assert not result.is_error
    assert "customer" in result.content[0].text

@pytest.mark.asyncio
async def test_query_database_invalid_sql():
    result = await query_database("DROP TABLE customers")
    assert result.is_error

import logging

logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('mcp.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)