Design and implement Model Context Protocol servers for standardized AI-to-data integration with resources, tools, prompts, and security best practices
Master the Model Context Protocol (MCP) to build standardized, reusable integrations between AI systems and data sources, eliminating the N×M integration problem.
Open standard (November 2024, Anthropic) for connecting AI systems to external data sources and tools through a unified protocol.
The Problem: N agents × M tools = N×M custom integrations The Solution: N agents + M MCP servers = N+M integrations (any agent uses any tool)
┌─────────────┐
│ MCP Host │ (Claude Desktop, IDEs, Apps)
│ ┌─────┐ │
│ │Client│──┼──┐
│ └─────┘ │ │
└─────────────┘ │
│ JSON-RPC 2.0
│
┌────────────────┼─────────────┐
│ MCP Server ▼ │
│ ┌──────────────────┐ │
│ │ Resources │ │
│ │ Tools │ │
│ │ Prompts │ │
│ └──────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────┐ │
│ │ Data Source │ │
│ │ (DB, API, Files) │ │
│ └──────────────────┘ │
└─────────────────────────────┘
Purpose: Expose data for AI to read
Examples:
Definition:
{
"resources": [
{
"uri": "file:///docs/api-spec.md",
"name": "API Specification",
"mimeType": "text/markdown"
},
{
"uri": "db://customers/12345",
"name": "Customer Record",
"mimeType": "application/json"
}
]
}
Purpose: Functions AI can invoke
Examples:
Definition:
{
"tools": [
{
"name": "query_database",
"description": "Execute SQL query on customer database",
"inputSchema": {
"type": "object",
"properties": {
"query": {"type": "string"}
},
"required": ["query"]
}
}
]
}
Purpose: Reusable prompt templates
Examples:
Definition:
{
"prompts": [
{
"name": "analyze_customer",
"description": "Analyze customer behavior and generate insights",
"arguments": [
{
"name": "customer_id",
"description": "Customer identifier",
"required": true
}
]
}
]
}
from mcp import Server, Tool, Resource
server = Server("customer-data")
@server.resource("customer://")
async def get_customer(uri: str):
"""Expose customer data as resources"""
customer_id = uri.split("://")[1]
return {
"uri": uri,
"mimeType": "application/json",
"text": json.dumps(get_customer_data(customer_id))
}
@server.tool()
async def query_customers(
filters: dict
) -> list:
"""Query customer database"""
return database.query("customers", filters)
@server.prompt()
async def customer_analysis(customer_id: str):
"""Generate customer analysis prompt"""
return {
"messages": [
{
"role": "user",
"content": f"Analyze customer {customer_id} behavior and provide insights"
}
]
}
if __name__ == "__main__":
server.run()
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
const server = new Server({
name: "github-server",
version: "1.0.0"
}, {
capabilities: {
resources: {},
tools: {},
prompts: {}
}
});
server.setRequestHandler(ListResourcesRequestSchema, async () => {
return {
resources: [
{
uri: "github://issues",
name: "GitHub Issues",
mimeType: "application/json"
}
]
};
});
server.setRequestHandler(ListToolsRequestSchema, async () => {
return {
tools: [
{
name: "create_issue",
description: "Create a new GitHub issue",
inputSchema: {
type: "object",
properties: {
title: { type: "string" },
body: { type: "string" }
}
}
}
]
};
});
const transport = new StdioServerTransport();
await server.connect(transport);
# Via npm
npx @modelcontextprotocol/server-github
# Via Docker
docker run mcp-postgres-server
# Via Python
pip install mcp-server-slack
python -m mcp_server_slack
{
"mcpServers": {
"github": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-github"],
"env": {
"GITHUB_TOKEN": "your-token"
}
},
"postgres": {
"command": "docker",
"args": ["run", "mcp-postgres-server"],
"env": {
"DATABASE_URL": "postgresql://..."
}
}
}
}
from anthropic import Anthropic
client = Anthropic()
response = client.messages.create(
model="claude-sonnet-4-5",
mcp_servers={
"github": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-github"]
}
},
messages=[{
"role": "user",
"content": "List my GitHub issues"
}]
)
# OAuth 2.0 with Resource Indicators (RFC 8707)
server = Server(
"secure-api",
auth_type="oauth2",
scopes=["read:data", "write:data"]
)
@server.tool(required_scope="write:data")
async def update_record(record_id: str, data: dict):
# Only callable with write permissions
pass
@server.tool()
async def execute_query(query: str):
# Validate to prevent injection
if not is_safe_query(query):
raise ValueError("Unsafe query detected")
# Sanitize inputs
safe_query = sanitize_sql(query)
return database.execute(safe_query)
from functools import lru_cache
from time import time
@server.tool()
@rate_limit(calls=10, period=60) # 10 calls per minute
async def expensive_operation():
pass
@server.tool()
async def sensitive_operation(data: dict):
audit_log.write({
"timestamp": datetime.now(),
"operation": "sensitive_operation",
"user": current_user(),
"data": data
})
return process(data)
@server.resource("aggregated://customer")
async def aggregate_customer_data(customer_id: str):
"""Combine data from multiple sources"""
crm_data = await crm_server.get_resource(f"crm://{customer_id}")
support_data = await support_server.get_resource(f"support://{customer_id}")
analytics_data = await analytics_server.get_resource(f"analytics://{customer_id}")
return {
"uri": f"aggregated://customer/{customer_id}",
"data": {
**crm_data,
**support_data,
**analytics_data
}
}
from functools import lru_cache
@server.resource("cached://")
@lru_cache(maxsize=1000)
async def cached_resource(uri: str):
"""Cache frequently accessed resources"""
return await expensive_fetch(uri)
@server.tool()
async def stream_large_dataset(query: str):
"""Stream results for large datasets"""
async for chunk in database.stream(query):
yield chunk
from prometheus_client import Counter, Histogram
tool_calls = Counter('mcp_tool_calls', 'Tool invocations', ['tool_name'])
latency = Histogram('mcp_latency', 'Operation latency')
@server.tool()
@latency.time()
async def monitored_tool():
tool_calls.labels(tool_name='monitored_tool').inc()
# Tool implementation
import logging
logger = logging.getLogger("mcp_server")
@server.tool()
async def error_tracked_tool():
try:
return await risky_operation()
except Exception as e:
logger.error(f"Tool failed: {e}", exc_info=True)
raise
import pytest
from mcp.testing import MockServer
@pytest.mark.asyncio
async def test_customer_tool():
server = MockServer()
result = await server.call_tool("get_customer", {"id": "123"})
assert result["customer_id"] == "123"
@pytest.mark.asyncio
async def test_full_workflow():
# Start test server
async with TestMCPServer() as server:
# Test resource access
resource = await server.get_resource("test://data")
assert resource is not None
# Test tool execution
result = await server.call_tool("process_data", {"input": "test"})
assert result["success"] == True
Build MCP Server when:
Use existing MCP Server when:
Official:
SDKs:
pip install mcpnpm install @modelcontextprotocol/sdkMCP is the universal standard for AI-to-data integration in 2025 and beyond.
npx skills add frankxai/MCP Architecture Expert下载完整 Skill 目录，包含 SKILL.md 及所有相关文件
Search for places (restaurants, cafes, etc.) via Google Places API proxy on localhost.
Interact with GitHub using the `gh` CLI. Use `gh issue`, `gh pr`, `gh run`, and `gh api` for issues, PRs, CI runs, and advanced queries.
Create or update AgentSkills. Use when designing, structuring, or packaging skills with scripts, references, and assets.
Start voice calls via the OpenClaw voice-call plugin.
Notion API for creating and managing pages, databases, and blocks.
Gemini CLI for one-shot Q&A, summaries, and generation.
Category:developer