Agent skill
mcp-integration-toolkit
Create, configure, and integrate MCP (Model Context Protocol) servers and clients. Use when building MCP servers with FastMCP (Python) or MCP SDK (Node/TypeScript), integrating external APIs, or creating custom tools/resources for LLMs. Handles server creation, client configuration, transport protocols, and security patterns.
Install this agent skill to your Project
npx add-skill https://github.com/majiayu000/claude-skill-registry/tree/main/skills/development/mcp-integration-toolkit
SKILL.md
MCP Integration Toolkit
Purpose
The Model Context Protocol (MCP) is the new standard for connecting LLMs to external tools, data sources, and APIs. This Skill provides comprehensive support for:
- Creating MCP Servers - Build servers in Python (FastMCP) or Node/TypeScript (MCP SDK)
- Client Integration - Configure Claude Code and other clients to use MCP servers
- Protocol Patterns - Implement tools, resources, and prompts correctly
- Security & Best Practices - Handle authentication, rate limiting, and error handling
- Testing & Debugging - Validate MCP implementations and troubleshoot issues
When to Use This Skill
- Building MCP servers to expose APIs or services to LLMs
- Integrating third-party services (databases, APIs, file systems) via MCP
- Creating custom tools for Claude Code or other MCP clients
- Debugging MCP server/client communication issues
- Converting existing tools to MCP protocol
- Setting up MCP transport layers (stdio, SSE, HTTP)
Core Concepts
MCP Architecture
┌─────────────┐ ┌──────────────┐ ┌────────────┐
│ │ │ │ │ │
│ LLM Client │◄────────┤ MCP Server │◄────────┤ External │
│ (Claude) │ MCP │ (FastMCP/ │ API │ Service │
│ │ Proto │ MCP SDK) │ Calls │ │
└─────────────┘ └──────────────┘ └────────────┘
MCP Components
- Tools - Functions the LLM can call (like API endpoints)
- Resources - Data the LLM can read (files, database records, API responses)
- Prompts - Pre-configured templates the LLM can use
- Transports - Communication layers (stdio, Server-Sent Events, HTTP)
Knowledge Resources
Official Documentation
- MCP Specification - Protocol specification
- FastMCP (Python) - Python framework for MCP servers
- MCP SDK (TypeScript) - Official TypeScript/Node.js SDK
- MCP Documentation - Official guides and tutorials
Key Patterns
FastMCP Server (Python):
from fastmcp import FastMCP
mcp = FastMCP("my-server")
@mcp.tool()
def search_database(query: str) -> dict:
"""Search the database for records matching query"""
# Implementation
return {"results": [...]}
@mcp.resource("config://settings")
def get_settings() -> str:
"""Return current configuration"""
return json.dumps(settings)
if __name__ == "__main__":
mcp.run()
MCP SDK Server (TypeScript):
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
const server = new Server({
name: "my-server",
version: "1.0.0",
});
server.setRequestHandler(ListToolsRequestSchema, async () => ({
tools: [{
name: "search_database",
description: "Search the database",
inputSchema: { type: "object", properties: { query: { type: "string" } } }
}]
}));
const transport = new StdioServerTransport();
await server.connect(transport);
Claude Code Configuration (~/.config/claude-code/mcp.json):
{
"mcpServers": {
"my-server": {
"command": "python",
"args": ["/path/to/server.py"],
"env": {
"API_KEY": "${API_KEY}"
}
}
}
}
Common MCP Gotchas
- Transport Mismatch - Server and client must use compatible transports (stdio most common for local)
- Schema Validation - Tool input schemas must be valid JSON Schema (not TypeScript types)
- Stdio Buffering - Use
sys.stdout.flush()in Python or avoid logging to stdout when using stdio transport - Authentication - Handle API keys via environment variables, never hardcode in server
- Error Handling - Return proper error responses, don't raise unhandled exceptions
- Async Operations - FastMCP handles sync/async automatically, MCP SDK requires explicit async/await
- Resource URIs - Must follow URI format:
protocol://path(e.g.,file:///path,db://table/id) - Tool Discovery - LLM clients cache tool lists, restart client after server changes
Implementation Patterns
Pattern 1: API Integration Server
Expose external API to LLMs with rate limiting and caching:
from fastmcp import FastMCP
import httpx
from functools import lru_cache
from datetime import datetime, timedelta
mcp = FastMCP("api-integration")
# Rate limiting
last_call = {}
RATE_LIMIT_SECONDS = 1
@mcp.tool()
async def call_external_api(endpoint: str, params: dict = None) -> dict:
"""Call external API with rate limiting"""
now = datetime.now()
if endpoint in last_call:
elapsed = (now - last_call[endpoint]).total_seconds()
if elapsed < RATE_LIMIT_SECONDS:
raise ValueError(f"Rate limit: wait {RATE_LIMIT_SECONDS - elapsed:.1f}s")
async with httpx.AsyncClient() as client:
response = await client.get(f"https://api.example.com/{endpoint}", params=params)
response.raise_for_status()
last_call[endpoint] = now
return response.json()
Pattern 2: Database Resource Server
Expose database tables as MCP resources:
from fastmcp import FastMCP
import sqlite3
mcp = FastMCP("database-server")
@mcp.resource("db://users/{user_id}")
def get_user(user_id: str) -> str:
"""Get user by ID"""
conn = sqlite3.connect("database.db")
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))
row = cursor.fetchone()
conn.close()
return json.dumps(dict(row)) if row else None
@mcp.tool()
def list_users(limit: int = 10) -> list[dict]:
"""List users from database"""
conn = sqlite3.connect("database.db")
cursor = conn.cursor()
cursor.execute("SELECT * FROM users LIMIT ?", (limit,))
rows = cursor.fetchall()
conn.close()
return [dict(row) for row in rows]
Pattern 3: File System Server
Provide safe file system access with sandboxing:
from fastmcp import FastMCP
from pathlib import Path
mcp = FastMCP("filesystem-server")
ALLOWED_PATHS = [Path("/safe/directory")]
def is_safe_path(path: str) -> bool:
"""Check if path is within allowed directories"""
p = Path(path).resolve()
return any(p.is_relative_to(allowed) for allowed in ALLOWED_PATHS)
@mcp.resource("file://{path}")
def read_file(path: str) -> str:
"""Read file contents (sandboxed)"""
if not is_safe_path(path):
raise PermissionError(f"Access denied: {path}")
return Path(path).read_text()
@mcp.tool()
def list_directory(path: str) -> list[str]:
"""List directory contents (sandboxed)"""
if not is_safe_path(path):
raise PermissionError(f"Access denied: {path}")
return [str(p) for p in Path(path).iterdir()]
Testing & Debugging
Testing MCP Servers Locally
Test with MCP Inspector (official debugging tool):
npm install -g @modelcontextprotocol/inspector
mcp-inspector python server.py
Test with Claude Code:
- Add server to
~/.config/claude-code/mcp.json - Restart Claude Code
- Use
/mcpcommand to list available tools - Test tool invocation
Unit Testing (FastMCP):
import pytest
from server import mcp
def test_search_tool():
result = mcp.tools["search_database"]("test query")
assert "results" in result
assert isinstance(result["results"], list)
Debugging Checklist
- Server Starts - Run server directly, check for startup errors
- Transport Working - Verify stdio/SSE/HTTP transport responds
- Schema Valid - Validate JSON schemas with online validator
- Tools Discovered - Check client can list tools
- Tool Execution - Test tool with sample inputs
- Error Handling - Verify errors return proper format, don't crash server
- Authentication - Test with/without credentials
- Logging - Use stderr for logs (stdout reserved for protocol)
Security Best Practices
1. Authentication & Authorization
import os
from fastmcp import FastMCP
mcp = FastMCP("secure-server")
API_KEY = os.getenv("API_KEY")
@mcp.tool()
def protected_operation(auth_token: str, data: dict) -> dict:
"""Operation requiring authentication"""
if auth_token != API_KEY:
raise PermissionError("Invalid authentication")
# Perform operation
return {"status": "success"}
2. Input Validation
from pydantic import BaseModel, Field, validator
class SearchParams(BaseModel):
query: str = Field(..., min_length=1, max_length=500)
limit: int = Field(10, ge=1, le=100)
@validator("query")
def sanitize_query(cls, v):
# Remove SQL injection attempts
dangerous = ["DROP", "DELETE", "INSERT", "UPDATE"]
if any(word in v.upper() for word in dangerous):
raise ValueError("Invalid query")
return v
@mcp.tool()
def search(params: SearchParams) -> list:
"""Search with validated params"""
# params is guaranteed valid
return perform_search(params.query, params.limit)
3. Rate Limiting
from collections import defaultdict
from datetime import datetime, timedelta
class RateLimiter:
def __init__(self, calls_per_minute=60):
self.calls = defaultdict(list)
self.limit = calls_per_minute
def check(self, client_id: str) -> bool:
now = datetime.now()
minute_ago = now - timedelta(minutes=1)
# Clean old calls
self.calls[client_id] = [
t for t in self.calls[client_id] if t > minute_ago
]
if len(self.calls[client_id]) >= self.limit:
return False
self.calls[client_id].append(now)
return True
limiter = RateLimiter(calls_per_minute=60)
@mcp.tool()
def rate_limited_operation(client_id: str, data: dict) -> dict:
if not limiter.check(client_id):
raise ValueError("Rate limit exceeded")
# Perform operation
return {"status": "success"}
Migration Patterns
Converting Existing API to MCP
Before (Direct API):
@app.get("/api/search")
def search_endpoint(query: str):
return {"results": search_database(query)}
After (MCP Tool):
from fastmcp import FastMCP
mcp = FastMCP("search-server")
@mcp.tool()
def search(query: str) -> dict:
"""Search database and return results"""
return {"results": search_database(query)}
Converting Claude Code Tool to MCP
Before (Claude Code bash tool):
# User calls: claude "search database for X"
# Claude uses: Bash tool to run python script
After (MCP Tool):
# User calls: claude "search database for X"
# Claude uses: MCP tool directly
# Benefits: Type safety, better error handling, no shell injection
Advanced Topics
Multi-Transport Server
Support both stdio (local) and HTTP (remote):
from fastmcp import FastMCP
import os
mcp = FastMCP("multi-transport-server")
# Define tools...
if __name__ == "__main__":
transport = os.getenv("MCP_TRANSPORT", "stdio")
if transport == "stdio":
mcp.run() # Default stdio
elif transport == "http":
mcp.run_http(host="0.0.0.0", port=8000)
elif transport == "sse":
mcp.run_sse(host="0.0.0.0", port=8000)
Resource Subscriptions
Notify clients when resources change:
from fastmcp import FastMCP
mcp = FastMCP("subscription-server")
@mcp.resource("data://latest", notify_on_change=True)
def get_latest_data() -> str:
"""Get latest data (client receives updates)"""
return get_current_data()
# When data changes:
mcp.notify_resource_changed("data://latest")
Custom Prompts
Provide pre-configured prompt templates:
@mcp.prompt("code-review")
def code_review_prompt(language: str) -> str:
"""Generate code review prompt for specific language"""
return f"""You are reviewing {language} code. Focus on:
1. Code style and idioms for {language}
2. Performance patterns specific to {language}
3. Common {language} anti-patterns
4. Security issues relevant to {language}
Please provide detailed, actionable feedback."""
Production Deployment
Deployment Checklist
- Environment variables configured (API keys, database URLs)
- Logging to stderr (stdout reserved for MCP protocol)
- Error handling for all tools (return errors, don't crash)
- Rate limiting implemented
- Input validation for all parameters
- Authentication/authorization for sensitive operations
- Health check endpoint (if using HTTP transport)
- Monitoring and alerting setup
- Documentation for all tools and resources
- Version pinning for dependencies
Example Production Server
from fastmcp import FastMCP
import logging
import sys
from typing import Optional
# Configure logging to stderr
logging.basicConfig(
level=logging.INFO,
stream=sys.stderr,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
mcp = FastMCP("production-server", version="1.0.0")
@mcp.tool()
def production_tool(param: str) -> dict:
"""Production-ready tool with full error handling"""
try:
logger.info(f"Tool called with param: {param}")
# Validate input
if not param or len(param) > 1000:
raise ValueError("Invalid param length")
# Perform operation
result = perform_operation(param)
logger.info("Tool completed successfully")
return {"status": "success", "result": result}
except ValueError as e:
logger.error(f"Validation error: {e}")
return {"status": "error", "message": str(e)}
except Exception as e:
logger.exception("Unexpected error in production_tool")
return {"status": "error", "message": "Internal server error"}
if __name__ == "__main__":
logger.info("Starting MCP server")
mcp.run()
Next Steps
After creating an MCP server:
- Test Locally - Use MCP Inspector or Claude Code to validate
- Document Tools - Write clear descriptions for all tools/resources
- Add Examples - Provide usage examples in tool descriptions
- Configure Client - Add to
mcp.jsonfor Claude Code integration - Monitor Usage - Log tool calls and errors for debugging
- Iterate - Refine based on actual LLM usage patterns
Related Skills
git-mastery-suite- For managing MCP server code in Gitdeployment-automation-toolkit- For deploying MCP servers to productionsecurity-scanning-suite- For security analysis of MCP implementationsapi-integration-toolkit- For wrapping external APIs as MCP servers
References
- MCP Awesome List - Community MCP servers
- FastMCP Examples - Python examples
- MCP SDK Examples - TypeScript examples
- Claude Code MCP Guide - Integration guide
Recommended Agent Skills
Expand your agent's capabilities with these related and highly-rated skills.
agent-ops-spec
Manage specification documents in .agent/specs/. Use when user provides requirements, acceptance criteria, or feature descriptions that need to be tracked and validated against implementation.
agent-ops-state
Maintain .agent state files. Use at session start, after meaningful steps, and before concluding: read/update constitution/memory/focus/issues/baseline consistently.
agent-ops-spec
Manage specification documents in .agent/specs/. Use when user provides requirements, acceptance criteria, or feature descriptions that need to be tracked and validated against implementation.
agent-ops-testing
Test strategy, execution, and coverage analysis. Use when designing tests, running test suites, or analyzing test results beyond baseline checks.
agent-ops-testing
Test strategy, execution, and coverage analysis. Use when designing tests, running test suites, or analyzing test results beyond baseline checks.
agent-ops-state
Maintain .agent state files. Use at session start, after meaningful steps, and before concluding: read/update constitution/memory/focus/issues/baseline consistently.
Didn't find tool you were looking for?