| name | mcp |
| description | Comprehensive skill for Model Context Protocol (MCP) - building servers and clients for LLM context integration Use when this capability is needed. |
| metadata | {"author":"aeonbridge"} |
Model Context Protocol (MCP) Skill
Expert assistance for building MCP servers and clients to extend LLM capabilities with custom context, tools, and resources using the standardized Model Context Protocol.
When to Use This Skill
This skill should be used when:
- Building MCP servers to expose tools, resources, or prompts to LLMs
- Creating MCP clients to integrate with Claude or other LLM applications
- Extending Claude Desktop with custom functionality
- Implementing standardized LLM context protocols
- Developing plugins for AI applications
- Creating reusable AI tool integrations
- Building enterprise AI integrations with security
- Implementing OAuth 2.1 authorization for MCP
- Debugging MCP servers with the Inspector tool
- Questions about MCP architecture and best practices
- Integrating external APIs with LLMs
- Building knowledge bases accessible to AI assistants
Overview
What is MCP?
Model Context Protocol (MCP) is a standardized protocol that enables:
- LLMs to access external tools and data sources
- Bidirectional communication between AI applications and integrations
- Unified interface for context providers
- Secure, scalable AI integrations
Core Value:
"MCP standardizes how LLMs access context, enabling reusable integrations across applications."
MCP Architecture
┌─────────────────┐
│ LLM Client │ (Claude Desktop, Custom Client)
│ (MCP Client) │
└────────┬────────┘
│ MCP Protocol
│ (JSON-RPC)
┌────────▼────────┐
│ MCP Server │ (Your Integration)
└────────┬────────┘
│
┌────────▼────────┐
│ External APIs │ (Database, APIs, Files)
│ & Data Sources │
└─────────────────┘
Core Concepts
1. Resources: File-like data readable by clients
- Similar to GET endpoints
- Provide information without side effects
- Examples: file contents, API responses, database records
2. Tools: Functions callable by LLMs
- Similar to POST endpoints
- Execute code and produce side effects
- Require user approval
- Examples: send email, create file, call API
3. Prompts: Pre-written templates
- Reusable interaction patterns
- Consistent user-model interactions
- Examples: code review template, data analysis prompt
Installation
Python SDK
uv add "mcp[cli]"
pip install "mcp[cli]"
pip install "mcp[cli,dev]"
Requirements:
- Python 3.10+
- MCP SDK 1.2.0+
TypeScript SDK
npm install @modelcontextprotocol/sdk zod
npm install --save-dev typescript @types/node
Requirements:
MCP Inspector
npx @modelcontextprotocol/inspector <command>
Building MCP Servers
Python Server (FastMCP)
Basic Structure:
from mcp import FastMCP
mcp = FastMCP("my-server")
@mcp.resource("file://readme")
def get_readme() -> str:
"""Provide README contents"""
with open("README.md") as f:
return f.read()
@mcp.tool()
async def add_numbers(a: int, b: int) -> int:
"""Add two numbers together
Args:
a: First number
b: Second number
Returns:
Sum of a and b
"""
return a + b
@mcp.prompt()
def code_review_prompt(language: str = "python") -> list[dict]:
"""Generate code review prompt"""
return [
{
"role": "user",
"content": f"Review this {language} code for best practices"
}
]
Run Server:
uv run mcp dev my_server.py
python -m mcp.server.stdio my_server:mcp
python -m mcp.server.sse my_server:mcp --port 8000
TypeScript Server
Basic Structure:
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
const server = new Server(
{
name: "my-server",
version: "1.0.0",
},
{
capabilities: {
resources: {},
tools: {},
prompts: {},
},
}
);
server.setRequestHandler(
"tools/list",
async () => ({
tools: [
{
name: "add_numbers",
description: "Add two numbers",
inputSchema: z.object({
a: z.number().describe("First number"),
b: z.number().describe("Second number"),
}),
},
],
})
);
server.setRequestHandler(
"tools/call",
async (request) => {
if (request.params.name === "add_numbers") {
{ a, b } = request..;
{
: [
{
: ,
: (a + b),
},
],
};
}
();
}
);
transport = ();
server.(transport);
Build and Run:
npm run build
node build/index.js
Critical Best Practices
⚠️ NEVER write to stdout in STDIO servers!
print("Debug message")
import logging
logging.basicConfig(level=logging.INFO, handlers=[
logging.FileHandler('server.log')
])
logger = logging.getLogger(__name__)
logger.info("Debug message")
import sys
print("Debug message", file=sys.stderr)
console.log("Debug message");
console.error("Debug message");
import winston from 'winston';
const logger = winston.createLogger({
transports: [new winston.transports.File({ filename: 'server.log' })]
});
logger.info("Debug message");
Building MCP Clients
Python Client
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from anthropic import Anthropic
import asyncio
class MCPClient:
def __init__(self):
self.anthropic = Anthropic()
self.session = None
async def connect_to_server(self, server_script_path: str):
"""Connect to MCP server"""
server_params = StdioServerParameters(
command="python",
args=[server_script_path],
env=None
)
stdio_transport = await stdio_client(server_params)
self.stdio, self.write = stdio_transport
self.session = ClientSession(self.stdio, self.write)
await self.session.__aenter__()
await self.session.initialize()
async def list_tools(self):
"""Get available tools from server"""
response = await self.session.list_tools()
return response.tools
async def ():
tools = .list_tools()
claude_tools = [
{
: tool.name,
: tool.description,
: tool.inputSchema
}
tool tools
]
messages = [{: , : query}]
:
response = .anthropic.messages.create(
model=,
max_tokens=,
tools=claude_tools,
messages=messages
)
response.stop_reason != :
response.content[].text
content response.content:
content. == :
result = .session.call_tool(
content.name,
content.
)
messages.append({
: ,
: response.content
})
messages.append({
: ,
: [{
: ,
: content.,
: result.content
}]
})
():
client = MCPClient()
client.connect_to_server()
result = client.process_query()
(result)
__name__ == :
asyncio.run(main())
TypeScript Client
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
import Anthropic from "@anthropic-ai/sdk";
const transport = new StdioClientTransport({
command: "python",
args: ["server.py"],
});
const client = new Client(
{
name: "my-client",
version: "1.0.0",
},
{
capabilities: {},
}
);
await client.connect(transport);
const toolsResponse = await client.listTools();
const tools = toolsResponse.tools.map(tool => ({
name: tool.name,
description: tool.description,
input_schema: tool.inputSchema,
}));
const anthropic = new Anthropic({
apiKey: process.env.ANTHROPIC_API_KEY,
});
messages = [
{ : , : }
];
() {
response = anthropic..({
: ,
: ,
tools,
messages,
});
(response. !== ) {
.(response.[].);
;
}
( content response.) {
(content. === ) {
result = client.({
: content.,
: content.,
});
messages.({
: ,
: response.,
});
messages.({
: ,
: [{
: ,
: content.,
: result.,
}],
});
}
}
}
Integration with Claude Desktop
Configuration
Edit claude_desktop_config.json:
macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
Windows: %APPDATA%\Claude\claude_desktop_config.json
{
"mcpServers": {
"my-server": {
"command": "python",
"args": ["/absolute/path/to/server.py"]
},
"another-server": {
"command": "node",
"args": ["/absolute/path/to/server.js"]
}
}
}
Important:
- Use absolute paths only
- Fully restart Claude Desktop (not just close window)
- Check logs at
~/Library/Logs/Claude/ (macOS)
Verification
- Restart Claude Desktop completely
- Start new conversation
- Look for 🔨 icon indicating tools are available
- Tools should appear in tool selection menu
Security and Authorization
OAuth 2.1 Implementation
MCP uses standardized OAuth 2.1 for authorization:
When to Use:
- Servers handle sensitive data
- Enterprise environments
- Multi-user systems
- Audit trail requirements
Authorization Flow:
from mcp.server import Server
from mcp.server.auth import OAuth2Provider
oauth_provider = OAuth2Provider(
authorization_endpoint="https://auth.example.com/oauth/authorize",
token_endpoint="https://auth.example.com/oauth/token",
client_id="your-client-id",
client_secret="your-client-secret",
scopes=["read:data", "write:data"]
)
server = Server("secure-server", auth=oauth_provider)
@server.tool(required_scopes=["write:data"])
async def write_file(path: str, content: str):
"""Write file (requires write scope)"""
if not server.verify_scope("write:data"):
raise PermissionError("Insufficient permissions")
with open(path, 'w') as f:
f.write(content)
return "File written"
Security Best Practices
import os
from typing import Optional
API_KEY = os.getenv("API_KEY")
if not API_KEY:
raise ValueError("API_KEY environment variable required")
def validate_token(token: str) -> bool:
try:
payload = jwt.decode(
token,
PUBLIC_KEY,
algorithms=["RS256"],
audience="your-resource-server"
)
return True
except jwt.InvalidTokenError:
return False
ACCESS_TOKEN_LIFETIME = 900
TOOL_SCOPES = {
"read_file": ["read:files"],
"write_file": ["write:files"],
"delete_file": ["write:files", "delete:files"]
}
try:
result = perform_action()
except Exception as e:
logger.error(f"Action failed: ")
{: }
is_localhost() using_https():
SecurityError()
MCP Inspector
Running Inspector
npx @modelcontextprotocol/inspector python server.py
npx @modelcontextprotocol/inspector python server.py --arg value
npx @modelcontextprotocol/inspector node server.js
npx @modelcontextprotocol/inspector npx server-package
Inspector Features
Server Connection Panel:
- Configure transport methods (stdio, HTTP, SSE)
- Customize command-line arguments
- Test connection status
Resources Tab:
- View all available resources
- Test resource subscriptions
- Preview resource content
- Check MIME types and metadata
Prompts Tab:
- List prompt templates
- Test with custom arguments
- Preview generated messages
Tools Tab:
- View tool schemas and descriptions
- Execute tools with custom inputs
- Test parameter validation
- Check response formats
Notifications Pane:
- Monitor server logs
- View real-time notifications
- Track errors and warnings
Debugging Workflow
npx @modelcontextprotocol/inspector python server.py
Common Patterns
File System Server
from mcp import FastMCP
from pathlib import Path
import os
mcp = FastMCP("filesystem")
@mcp.tool()
async def read_file(path: str) -> str:
"""Read file contents
Args:
path: Path to file
Returns:
File contents
"""
file_path = Path(path)
if not file_path.exists():
raise FileNotFoundError(f"File not found: {path}")
return file_path.read_text()
@mcp.tool()
async def write_file(path: str, content: str) -> str:
"""Write content to file
Args:
path: Path to file
content: Content to write
Returns:
Success message
"""
file_path = Path(path)
file_path.parent.mkdir(parents=True, exist_ok=True)
file_path.write_text(content)
return f"Wrote {len(content)} bytes to {path}"
@mcp.tool()
async def list_directory(path: str) -> list[str]:
"""List directory contents
Args:
path: Directory path
Returns:
List of filenames
"""
dir_path = Path(path)
dir_path.is_dir():
NotADirectoryError()
[item.name item dir_path.iterdir()]
API Integration Server
from mcp import FastMCP
import httpx
import os
mcp = FastMCP("api-server")
API_KEY = os.getenv("API_KEY")
BASE_URL = "https://api.example.com"
@mcp.tool()
async def search_api(query: str, limit: int = 10) -> dict:
"""Search API
Args:
query: Search query
limit: Maximum results
Returns:
Search results
"""
async with httpx.AsyncClient() as client:
response = await client.get(
f"{BASE_URL}/search",
params={"q": query, "limit": limit},
headers={"Authorization": f"Bearer {API_KEY}"}
)
response.raise_for_status()
return response.json()
@mcp.resource("api://status")
async def api_status() -> dict:
"""Get API status"""
async with httpx.AsyncClient() as client:
response = await client.get(f"{BASE_URL}/status")
return response.json()
Database Server
from mcp import FastMCP
import sqlite3
from typing import List, Dict
mcp = FastMCP("database")
DB_PATH = "data.db"
@mcp.tool()
async def query_database(sql: str) -> List[Dict]:
"""Execute SQL query
Args:
sql: SQL query (SELECT only)
Returns:
Query results
"""
if not sql.strip().upper().startswith("SELECT"):
raise ValueError("Only SELECT queries allowed")
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
try:
cursor.execute(sql)
results = [dict(row) for row in cursor.fetchall()]
return results
finally:
conn.close()
@mcp.tool()
async def insert_record(table: str, data: dict) -> str:
"""Insert record into table
Args:
table: Table name
data: Column-value pairs
Returns:
Success message
"""
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
columns = ", ".join(data.keys())
placeholders = ", ".join(["?" for _ in data])
sql = f"INSERT INTO () VALUES ()"
:
cursor.execute(sql, (data.values()))
conn.commit()
:
conn.close()
Advanced Features
Structured Output
from mcp import FastMCP
from pydantic import BaseModel
from typing import List
mcp = FastMCP("structured-server")
class SearchResult(BaseModel):
title: str
url: str
snippet: str
score: float
class SearchResponse(BaseModel):
query: str
results: List[SearchResult]
total: int
@mcp.tool()
async def search(query: str) -> SearchResponse:
"""Search with structured output"""
results = [
SearchResult(
title="Result 1",
url="https://example.com/1",
snippet="First result",
score=0.95
),
SearchResult(
title="Result 2",
url="https://example.com/2",
snippet="Second result",
score=0.87
)
]
return SearchResponse(
query=query,
results=results,
total=len(results)
)
Context and Progress
from mcp import FastMCP, Context
import asyncio
mcp = FastMCP("progress-server")
@mcp.tool()
async def long_running_task(ctx: Context, iterations: int) -> str:
"""Long running task with progress
Args:
ctx: MCP context (auto-injected)
iterations: Number of iterations
Returns:
Completion message
"""
for i in range(iterations):
await ctx.report_progress(
progress=i,
total=iterations,
message=f"Processing {i}/{iterations}"
)
ctx.logger.info(f"Iteration {i}")
await asyncio.sleep(0.1)
return f"Completed {iterations} iterations"
Lifespan Management
from mcp import FastMCP
from contextlib import asynccontextmanager
import httpx
@asynccontextmanager
async def lifespan(app):
print("Starting server...")
app.http_client = httpx.AsyncClient()
yield
print("Shutting down server...")
await app.http_client.aclose()
mcp = FastMCP("lifecycle-server", lifespan=lifespan)
@mcp.tool()
async def fetch_url(ctx: Context, url: str) -> str:
"""Fetch URL using shared HTTP client
Args:
ctx: MCP context
url: URL to fetch
Returns:
Response text
"""
http_client = ctx.lifespan.http_client
response = await http_client.get(url)
return response.text
Transport Options
STDIO (Standard Input/Output)
Best For: Claude Desktop integration, CLI tools
python -m mcp.server.stdio server:mcp
node build/index.js
SSE (Server-Sent Events)
Best For: Web applications, HTTP-based clients
python -m mcp.server.sse server:mcp --port 8000
python -m mcp.server.sse server:mcp --port 8000 --cors-origin "*"
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
const transport = new SSEServerTransport("/sse", res);
await server.connect(transport);
HTTP (Streamable)
Best For: Modern web applications, recommended
mcp = FastMCP("server", streamable_http=True)
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/http.js";
const transport = new StreamableHTTPServerTransport({
endpoint: "/mcp",
});
await server.connect(transport);
Troubleshooting
Common Issues
Server not appearing in Claude Desktop:
cat ~/Library/Application\ Support/Claude/claude_desktop_config.json | python -m json.tool
tail -f ~/Library/Logs/Claude/mcp*.log
STDIO Communication Broken:
print("Debug message")
import logging
logging.basicConfig(
level=logging.DEBUG,
filename='server.log',
format='%(asctime)s - %(levelname)s - %(message)s'
)
Tool Not Executing:
@mcp.tool()
async def my_tool(
param1: str,
param2: int = 10
) -> str:
"""Tool description must be clear
Args:
param1: First parameter description
param2: Second parameter description (optional)
Returns:
What this tool returns
"""
pass
Connection Timeout:
import httpx
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.get(url)
Best Practices
1. Clear Documentation
@mcp.tool()
async def process_data(
data: str,
format: str = "json",
validate: bool = True
) -> dict:
"""Process and validate data
This tool processes input data and optionally validates it
against a schema before returning the processed result.
Args:
data: Raw input data to process
format: Output format (json, xml, csv)
validate: Whether to validate against schema
Returns:
Processed data with metadata
Raises:
ValueError: If data format is invalid
ValidationError: If validation fails
Examples:
>>> process_data('{"key": "value"}', format="json")
{"processed": true, "data": {...}}
"""
pass
2. Error Handling
@mcp.tool()
async def safe_operation(path: str) -> str:
"""Safe operation with proper error handling"""
try:
if not path:
raise ValueError("Path cannot be empty")
result = perform_operation(path)
return f"Success: {result}"
except FileNotFoundError:
return "Error: File not found"
except PermissionError:
return "Error: Permission denied"
except Exception as e:
logger.error(f"Operation failed: {e}", exc_info=True)
return "Error: Operation failed"
3. Input Validation
from pydantic import BaseModel, Field, validator
class FileOperation(BaseModel):
path: str = Field(..., min_length=1, max_length=255)
content: str = Field(..., max_length=1000000)
@validator('path')
def validate_path(cls, v):
if '..' in v or v.startswith('/'):
raise ValueError("Invalid path")
return v
@mcp.tool()
async def write_validated(operation: FileOperation) -> str:
"""Write file with validated input"""
return write_file(operation.path, operation.content)
4. Testing
import pytest
from mcp.testing import MCPTestClient
@pytest.mark.asyncio
async def test_add_numbers():
"""Test add_numbers tool"""
async with MCPTestClient(mcp) as client:
tools = await client.list_tools()
assert "add_numbers" in [t.name for t in tools.tools]
result = await client.call_tool(
"add_numbers",
{"a": 5, "b": 3}
)
assert result.content[0].text == "8"
Resources
Official Documentation
Tools
- MCP Inspector: Debug and test servers interactively
- Claude Desktop: Reference implementation client
Community
License
MCP SDKs are licensed under MIT.
Note: This skill provides comprehensive guidance for building MCP servers and clients. Always follow security best practices and test thoroughly before production deployment.
Converted and distributed by TomeVault — claim your Tome and manage your conversions.