title: MCP Protocol for AI Tool Integration — Production Guide description: Model Context Protocol (MCP) implementation guide: server setup, client integration, tool sharing, security, and production deployment patterns. Original ops tutorial. tags: [ai, mcp, protocol, tools, ops] lang: en date: 2026-10-23 original: true faq: - q: "What is MCP and why do I need it?" a: "MCP (Model Context Protocol) is an open standard for connecting AI assistants to external data sources and tools. It provides a unified interface for LLMs to access databases, APIs, and services — replacing ad-hoc integrations with a standardized protocol." - q: "How to choose between stdio and SSE transport?" a: "stdio: local processes, simple debugging, single-user. SSE: remote servers, multi-client, production deployments. Use stdio for development/local tools, SSE for shared/proxied access." - q: "How to secure MCP servers in production?" a: "Always use TLS for SSE, implement authentication (JWT/API keys), sandbox tool execution, rate limit requests, and audit all tool calls. Never expose MCP servers directly to the internet without a reverse proxy." - q: "Can I share MCP tools across multiple AI clients?" a: "Yes — that's a key benefit. One MCP server can serve multiple clients (Claude, Cursor, custom apps). Use SSE transport with a central server for cross-client tool sharing." - q: "How to debug MCP connection issues?" a: "Enable verbose logging (--verbose flag), check JSON-RPC messages, verify server health endpoint, test with MCP Inspector tool, and check network connectivity between client and server."
MCP Protocol for AI Tool Integration — Production Guide
Use Case: Connect AI assistants (Claude, custom LLMs) to external tools and data sources using the Model Context Protocol (MCP) — standardizing tool integration across your AI ecosystem.
1. What is MCP?
The Model Context Protocol (MCP) is an open standard developed by Anthropic that enables AI assistants to securely connect to external tools and data sources.
Key Concepts: - MCP Host: The AI application (Claude Desktop, Cursor, custom app) - MCP Client: Runs inside the host, manages connections - MCP Server: Provides tools, resources, and prompts to the client
┌─────────────────────────────────────────────────────────────┐
│ MCP Host │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Claude │ │ Cursor │ │ Custom App │ │
│ │ Desktop │ │ IDE │ │ │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
│ │ │ │ │
│ └──────────────────┼───────────────────┘ │
│ ▼ │
│ ┌─────────────────────┐ │
│ │ MCP Client │ │
│ │ (JSON-RPC over │ │
│ │ stdio/SSE) │ │
│ └──────────┬──────────┘ │
└────────────────────────────┼────────────────────────────────┘
│
┌────────────────┼────────────────┐
▼ ▼ ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Database │ │ API │ │ System │
│ MCP Server │ │ MCP Server │ │ MCP Server │
└──────────────┘ └──────────────┘ └──────────────┘
2. MCP Transport Protocols
2.1 Stdio (Standard Input/Output)
Best for: Local tools, development, single-user scenarios
{
"mcpServers": {
"filesystem": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/home/user/docs"]
}
}
}
2.2 SSE (Server-Sent Events)
Best for: Remote servers, multi-client, production
3. Building an MCP Server
3.1 Python MCP Server (FastMCP)
from fastmcp import FastMCP
import psycopg2
import os
# Create MCP server
mcp = FastMCP("postgres-tools")
@mcp.tool()
def query_database(sql: str, limit: int = 100) -> str:
"""Execute a read-only SQL query against the analytics database."""
conn = psycopg2.connect(
host=os.environ["DB_HOST"],
database="analytics",
user=os.environ["DB_USER"],
password=os.environ["DB_PASSWORD"]
)
try:
cur = conn.cursor()
cur.execute(sql, {"limit": limit})
rows = cur.fetchall()
columns = [desc[0] for desc in cur.description]
return format_table(columns, rows)
finally:
conn.close()
@mcp.tool()
def get_table_schema(table_name: str) -> str:
"""Get the schema definition for a database table."""
conn = psycopg2.connect(
host=os.environ["DB_HOST"],
database="analytics",
user=os.environ["DB_USER"],
password=os.environ["DB_PASSWORD"]
)
try:
cur = conn.cursor()
cur.execute("""
SELECT column_name, data_type, is_nullable
FROM information_schema.columns
WHERE table_name = %s
ORDER BY ordinal_position
""", (table_name,))
rows = cur.fetchall()
return "\n".join(f"{r[0]}: {r[1]} ({r[2]})" for r in rows)
finally:
conn.close()
if __name__ == "__main__":
mcp.run(transport="sse", host="0.0.0.0", port=3001)
3.2 TypeScript MCP Server
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { CallToolRequestSchema } from "@modelcontextprotocol/sdk/types.js";
const server = new Server(
{ name: "postgres-tools", version: "1.0.0" },
{ capabilities: { tools: {} } }
);
server.setRequestHandler(CallToolRequestSchema, async (request) => {
const { name, arguments: args } = request.params;
if (name === "query_database") {
const result = await executeQuery(args.sql, args.limit || 100);
return { content: [{ type: "text", text: result }] };
}
throw new Error(`Unknown tool: ${name}`);
});
const transport = new StdioServerTransport();
await server.connect(transport);
4. MCP Client Integration
4.1 Claude Desktop Configuration
{
"mcpServers": {
"postgres": {
"command": "python",
"args": ["/opt/mcp/postgres_server.py"],
"env": {
"DB_HOST": "prod-db.internal",
"DB_USER": "mcp_reader",
"DB_PASSWORD": "${VAULT_DB_PASSWORD}"
}
},
"prometheus": {
"url": "http://mcp-prometheus:3001/sse"
}
}
}
4.2 Custom MCP Client (Python)
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def query_with_mcp():
server_params = StdioServerParameters(
command="python",
args=["/opt/mcp/postgres_server.py"],
env={**os.environ, "DB_HOST": "prod-db.internal"}
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# List available tools
tools = await session.list_tools()
# Call a tool
result = await session.call_tool(
"query_database",
{"sql": "SELECT count(*) FROM events WHERE date > '2026-10-01'", "limit": 1}
)
print(result.content[0].text)
5. Security Considerations
5.1 Tool Permissions
ALLOWED_SQL_PATTERNS = [
r"^SELECT\b", # Only SELECT queries
r"^SHOW\b", # SHOW commands
r"^DESCRIBE\b", # DESCRIBE
]
DENIED_OPERATIONS = [
"DROP", "ALTER", "DELETE", "INSERT", "UPDATE", "TRUNCATE"
]
def validate_sql(sql: str) -> bool:
"""Validate SQL for safety"""
sql_upper = sql.upper().strip()
for op in DENIED_OPERATIONS:
if op in sql_upper:
return False
for pattern in ALLOWED_SQL_PATTERNS:
if re.match(pattern, sql_upper):
return True
return False
5.2 Authentication and Authorization
from fastapi import Depends, HTTPException
from jose import jwt
SECRET_KEY = os.environ["MCP_JWT_SECRET"]
def authenticate(request: Request) -> dict:
auth_header = request.headers.get("Authorization")
if not auth_header or not auth_header.startswith("Bearer "):
raise HTTPException(401, "Missing token")
token = auth_header[7:]
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
return payload
except jwt.ExpiredSignatureError:
raise HTTPException(401, "Token expired")
except jwt.InvalidTokenError:
raise HTTPException(401, "Invalid token")
5.3 Rate Limiting
from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
@mcp.tool()
@limiter.limit("30/minute")
def query_database(sql: str, limit: int = 100) -> str:
"""Execute a read-only SQL query (rate-limited)."""
...
6. Production Deployment
6.1 Docker Compose for MCP Servers
version: "3.8"
services:
mcp-postgres:
build: ./mcp-servers/postgres
environment:
- DB_HOST=prod-db.internal
- DB_USER=mcp_reader
- DB_PASSWORD_FILE=/run/secrets/db_password
secrets:
- db_password
networks:
- mcp-internal
restart: unless-stopped
mcp-prometheus:
build: ./mcp-servers/prometheus
environment:
- PROMETHEUS_URL=http://prometheus:9090
networks:
- mcp-internal
restart: unless-stopped
secrets:
db_password:
file: ./secrets/db_password.txt
networks:
mcp-internal:
driver: bridge
internal: true
6.2 Health Checks
@app.get("/health")
async def health_check():
"""MCP server health endpoint"""
checks = {
"database": await check_db_connection(),
"memory_usage": psutil.virtual_memory().percent,
"uptime": time.time() - start_time
}
healthy = all(c.get("ok", False) for c in checks.values() if isinstance(c, dict))
status = 200 if healthy else 503
return JSONResponse(content=checks, status_code=status)
async def check_db_connection() -> dict:
try:
conn = psycopg2.connect(dbname="analytics", user="mcp_reader", host="prod-db.internal")
conn.close()
return {"ok": True, "latency_ms": 5}
except Exception as e:
return {"ok": False, "error": str(e)}
7. Debugging MCP Issues
7.1 Enable Verbose Logging
7.2 MCP Inspector Tool
# Use the official MCP Inspector for debugging
npx @modelcontextprotocol/inspector python postgres_server.py
7.3 Common Issues
| Symptom | Cause | Fix |
|---|---|---|
| Connection refused | Server not running | Check process status, verify port |
| Timeout | Network/firewall | Check connectivity, increase timeout |
| Auth error | Invalid token | Verify JWT secret, check expiration |
| Tool not found | Wrong tool name | List available tools with list_tools |
| SQL error | Permission denied | Check database user permissions |