
Mcp Integration Expert
- 25 installs
- 1 repo stars
- Updated November 29, 2025
- manutej/crush-mcp-server
Helps with ai & agent building tasks.
About
mcp-integration-expert is a Claude Code skill for ai & agent building. It helps solo builders move faster with AI-assisted development.
- mcp-integration-expert
- AI & Agent Building
- AI-coding skill
Mcp Integration Expert by the numbers
- 25 all-time installs (skills.sh)
- Ranked #9,796 of 16,546 AI & Agent Building skills by installs in the Skillselion catalog
- Data as of Aug 2, 2026 (Skillselion catalog sync)
npx skills add https://github.com/manutej/crush-mcp-server --skill mcp-integration-expertAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 25 |
|---|---|
| repo stars | ★ 1 |
| Last updated | November 29, 2025 |
| Repository | manutej/crush-mcp-server ↗ |
What it does
Helps with ai & agent building tasks.
Files
MCP Integration Expert
Comprehensive skill for researching, documenting, and integrating Model Context Protocol (MCP) servers and tools into Claude Code and AI applications.
When to Use This Skill
Use this skill when you need to:
- Research and evaluate MCP servers for integration
- Build custom MCP servers or clients
- Integrate existing MCP tools into Claude Code
- Document MCP server capabilities and usage patterns
- Troubleshoot MCP integration issues
- Implement MCP security and authentication patterns
- Create multi-language MCP implementations (Python, TypeScript, C#, Java, Rust)
- Design MCP-based AI agent workflows
- Evaluate MCP server trust scores and documentation quality
Model Context Protocol (MCP) Overview
What is MCP?
The Model Context Protocol (MCP) is an open standard introduced by Anthropic in November 2024 that standardizes how AI applications and Large Language Models (LLMs) integrate with external data sources, tools, and systems.
Key Analogy: MCP is like a "USB-C port for AI" - providing a universal, standardized interface for connecting AI models to diverse data sources and tools.
Core Architecture
┌─────────────────────────────────────────────────────────────┐
│ MCP Client │
│ (AI Application) │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Claude │ │ ChatGPT │ │ Custom │ │
│ │ Code │ │ Desktop │ │ App │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────┘
│
│ MCP Protocol
│
┌─────────────────────────────────────────────────────────────┐
│ MCP Servers │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Linear │ │ Postgres │ │ GitHub │ │ Custom │ │
│ │ Server │ │ Server │ │ Server │ │ Server │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────────────────┘MCP Protocol Communication Flow
sequenceDiagram
autonumber
actor User as 👤 User
participant ClientApp as 🖥️ Client App
participant ClientLLM as 🧠 Client LLM
participant Server1 as 🔧 MCP Server 1
participant Server2 as 📚 MCP Server 2
%% Discovery Phase
rect rgb(220, 240, 255)
Note over ClientApp, Server2: TOOL DISCOVERY PHASE
ClientApp->>+Server1: Request available tools/resources
Server1-->>-ClientApp: Return tool list (JSON)
ClientApp->>+Server2: Request available tools/resources
Server2-->>-ClientApp: Return tool list (JSON)
Note right of ClientApp: Store combined tool<br/>catalog locally
end
%% User Interaction
rect rgb(255, 240, 220)
Note over User, ClientLLM: USER INTERACTION PHASE
User->>+ClientApp: Enter natural language prompt
ClientApp->>+ClientLLM: Forward prompt + tool catalog
ClientLLM->>-ClientLLM: Analyze prompt & select tools
end
%% Tool Execution
rect rgb(220, 255, 220)
Note over ClientApp, Server1: TOOL EXECUTION PHASE
ClientLLM->>+ClientApp: Request tool execution
ClientApp->>+Server1: Execute specific tool
Server1-->>-ClientApp: Return results
ClientApp->>+ClientLLM: Process results
ClientLLM-->>-ClientApp: Generate response
ClientApp-->>-User: Display final answer
endMCP Core Primitives
MCP provides three core primitives for context exchange:
1. Resources: Expose data sources (files, databases, APIs) 2. Tools: Enable actions and operations 3. Prompts: Provide reusable prompt templates
MCP Integration Workflow
Phase 1: Research & Discovery
1.1 Identify MCP Server Needs
Questions to ask:
- What data sources do you need to access?
- What actions/tools do you need to perform?
- Are there existing MCP servers for your use case?
- Do you need to build a custom MCP server?
1.2 Research Available MCP Servers
Official MCP Server Repository: https://github.com/modelcontextprotocol
Popular MCP Servers (2025):
- Linear MCP: Project management and issue tracking
- Playwright MCP: Browser automation and testing
- Context7 MCP: Library documentation retrieval
- GitHub MCP: Repository management and automation
- Postgres MCP: Database queries and operations
- Google Drive MCP: File storage and retrieval
- Slack MCP: Team communication
- Stripe MCP: Payment processing
- Puppeteer MCP: Web scraping and automation
1.3 Evaluate MCP Server Quality
Trust Score Criteria (1-10 scale):
- 9-10: Highly authoritative (official SDKs, major platforms)
- 7-9: Well-maintained community projects
- 5-7: Experimental or niche implementations
- <5: Proof-of-concept or unmaintained
Documentation Coverage:
- High: 500+ code snippets
- Medium: 100-500 code snippets
- Low: <100 code snippets
Phase 2: MCP Server Architecture
2.1 MCP Server Components
// TypeScript MCP Server Structure
import { FastMCP } from '@modelcontextprotocol/typescript-sdk';
const mcp = new FastMCP({
name: "My MCP Server",
version: "1.0.0"
});
// 1. Resources: Expose data
mcp.resource("greeting://{name}", (name: string) => {
return `Hello, ${name}!`;
});
// 2. Tools: Enable actions
mcp.tool("calculate", {
description: "Perform calculations",
parameters: {
operation: { type: "string" },
a: { type: "number" },
b: { type: "number" }
}
}, async (params) => {
// Tool implementation
});
// 3. Prompts: Reusable templates
mcp.prompt("code-review", {
description: "Code review template",
arguments: ["language", "code"]
});2.2 Transport Mechanisms
MCP supports multiple transport mechanisms:
1. Stdio Transport (Default for local servers)
from mcp.server.fastmcp import FastMCP
from mcp.transports.stdio import serve_stdio
mcp = FastMCP("Demo")
# Run with stdio
asyncio.run(serve_stdio(mcp))2. HTTP/SSE Transport (For remote servers)
import { StreamableHTTPServerTransport } from '@modelcontextprotocol/typescript-sdk';
const transport = new StreamableHTTPServerTransport(8000);
await server.connect(transport);Phase 3: MCP Client Integration
3.1 Client Setup (Python)
from mcp.client import stdio_client, ClientSession
async def run_client():
server_params = {
"command": "python",
"args": ["server.py"]
}
async with stdio_client(server_params) as (reader, writer):
async with ClientSession(reader, writer) as session:
# Initialize connection
await session.initialize()
# Discover tools
tools = await session.list_tools()
# Call a tool
result = await session.call_tool("add", arguments={"a": 5, "b": 7})
print(f"Result: {result}")3.2 Client Setup (TypeScript)
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
class MCPClient {
private client: Client;
constructor() {
this.client = new Client({
name: "example-client",
version: "1.0.0"
}, {
capabilities: {
prompts: {},
resources: {},
tools: {}
}
});
}
async connectToServer(transport: Transport) {
await this.client.connect(transport);
}
async listTools() {
return await this.client.listTools();
}
async callTool(name: string, args: any) {
return await this.client.callTool({
name: name,
arguments: args
});
}
}3.3 Client Setup (C#)
using ModelContextProtocol.Client;
using ModelContextProtocol.Protocol.Transport;
var clientTransport = new StdioClientTransport(new()
{
Name = "Demo Server",
Command = "/path/to/server/executable",
Arguments = [],
});
await using var mcpClient = await McpClientFactory.CreateAsync(clientTransport);
// List tools
var tools = await mcpClient.ListToolsAsync();
foreach (var tool in tools)
{
Console.WriteLine($"Tool: {tool.Name}");
Console.WriteLine($"Description: {tool.Description}");
}
// Call a tool
var result = await mcpClient.CallToolAsync("add", new { a = 5, b = 7 });Phase 4: Building Custom MCP Servers
4.1 Python MCP Server (FastMCP)
from fastmcp import FastMCP
from fastmcp.transports.stdio import serve_stdio
import asyncio
mcp = FastMCP(
name="Weather MCP Server",
version="1.0.0"
)
@mcp.tool()
def get_weather(location: str) -> dict:
"""Gets current weather for a location."""
# Implementation would call weather API
return {
"temperature": 72.5,
"conditions": "Sunny",
"location": location
}
@mcp.resource("weather://{location}")
def weather_resource(location: str) -> str:
"""Get weather data as a resource"""
return f"Current weather in {location}"
class WeatherTools:
@mcp.tool()
def forecast(self, location: str, days: int = 1) -> dict:
"""Gets weather forecast"""
return {
"location": location,
"forecast": [
{"day": i+1, "temperature": 70 + i, "conditions": "Partly Cloudy"}
for i in range(days)
]
}
weather_tools = WeatherTools()
if __name__ == "__main__":
asyncio.run(serve_stdio(mcp))4.2 TypeScript MCP Server
import { FastMCP } from '@modelcontextprotocol/typescript-sdk';
import { serve_stdio } from '@modelcontextprotocol/typescript-sdk/transports/stdio';
const mcp = new FastMCP({
name: "Weather MCP Server",
version: "1.0.0"
});
mcp.tool("get_weather", {
description: "Gets current weather for a location",
parameters: {
location: { type: "string", required: true }
}
}, async (params) => {
return {
temperature: 72.5,
conditions: "Sunny",
location: params.location
};
});
mcp.resource("weather://{location}", async (location: string) => {
return `Current weather in ${location}`;
});
// Start server
serve_stdio(mcp);4.3 Java MCP Server
import io.modelcontextprotocol.server.McpServer;
import io.modelcontextprotocol.server.McpToolDefinition;
import io.modelcontextprotocol.server.transport.StdioServerTransport;
public class WeatherMcpServer {
public static void main(String[] args) throws Exception {
McpServer server = McpServer.builder()
.name("Weather MCP Server")
.version("1.0.0")
.build();
server.registerTool(McpToolDefinition.builder("weatherTool")
.description("Gets current weather for a location")
.parameter("location", String.class)
.execute((ctx) -> {
String location = ctx.getParameter("location", String.class);
WeatherData data = getWeatherData(location);
return ToolResponse.content(
String.format("Temperature: %.1f°F, Conditions: %s",
data.getTemperature(),
data.getConditions())
);
})
.build());
try (StdioServerTransport transport = new StdioServerTransport()) {
server.connect(transport);
Thread.currentThread().join();
}
}
}Phase 5: LLM Integration Patterns
5.1 Integrating MCP with OpenAI
import OpenAI from "openai";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
class MCPOpenAIClient {
private openai: OpenAI;
private mcpClient: Client;
constructor() {
this.openai = new OpenAI({
apiKey: process.env.OPENAI_API_KEY,
});
this.mcpClient = new Client({
name: "openai-mcp-client",
version: "1.0.0"
}, {
capabilities: { tools: {} }
});
}
// Adapt MCP tools to OpenAI format
mcpToolToOpenAITool(tool: any) {
return {
type: "function" as const,
function: {
name: tool.name,
description: tool.description,
parameters: {
type: "object",
properties: tool.input_schema.properties,
required: tool.input_schema.required,
},
},
};
}
async processRequest(userMessage: string) {
// Get MCP tools
const mcpTools = await this.mcpClient.listTools();
const openaiTools = mcpTools.map(t => this.mcpToolToOpenAITool(t));
// Call OpenAI with tools
const response = await this.openai.chat.completions.create({
model: "gpt-4",
messages: [{ role: "user", content: userMessage }],
tools: openaiTools,
});
// Handle tool calls
const toolCalls = response.choices[0].message.tool_calls;
if (toolCalls) {
for (const toolCall of toolCalls) {
const result = await this.mcpClient.callTool({
name: toolCall.function.name,
arguments: JSON.parse(toolCall.function.arguments),
});
console.log("Tool result:", result);
}
}
}
}5.2 Integrating MCP with Azure OpenAI (C#)
using Azure.AI.Inference;
using Azure;
using ModelContextProtocol.Client;
using System.Text.Json;
var endpoint = "https://models.inference.ai.azure.com";
var token = Environment.GetEnvironmentVariable("GITHUB_TOKEN");
var client = new ChatCompletionsClient(new Uri(endpoint), new AzureKeyCredential(token));
var mcpClient = await McpClientFactory.CreateAsync(clientTransport);
// Convert MCP tools to Azure format
ChatCompletionsToolDefinition ConvertFrom(string name, string description, JsonElement schema)
{
FunctionDefinition functionDefinition = new FunctionDefinition(name)
{
Description = description,
Parameters = BinaryData.FromObjectAsJson(new
{
Type = "object",
Properties = schema
})
};
return new ChatCompletionsToolDefinition(functionDefinition);
}
// Get tools from MCP server
var mcpTools = await mcpClient.ListToolsAsync();
var toolDefinitions = new List<ChatCompletionsToolDefinition>();
foreach (var tool in mcpTools)
{
JsonElement propertiesElement;
tool.JsonSchema.TryGetProperty("properties", out propertiesElement);
var def = ConvertFrom(tool.Name, tool.Description, propertiesElement);
toolDefinitions.Add(def);
}
// Use tools in chat completion
var chatHistory = new List<ChatRequestMessage>
{
new ChatRequestSystemMessage("You are a helpful assistant"),
new ChatRequestUserMessage("What's the weather in Seattle?")
};
var response = await client.CompleteAsync(chatHistory,
new ChatCompletionsOptions { Tools = toolDefinitions });Phase 6: MCP Configuration in Claude Code
6.1 Claude Code MCP Configuration
Claude Code automatically discovers and loads MCP servers configured in claude_desktop_config.json:
Location:
- macOS:
~/Library/Application Support/Claude/claude_desktop_config.json - Windows:
%APPDATA%\Claude\claude_desktop_config.json - Linux:
~/.config/Claude/claude_desktop_config.json
Configuration Format:
{
"mcpServers": {
"linear": {
"command": "npx",
"args": ["-y", "@linear/mcp-server"],
"env": {
"LINEAR_API_KEY": "your-api-key"
}
},
"playwright": {
"command": "npx",
"args": ["-y", "@playwright/mcp-server"]
},
"postgres": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-postgres"],
"env": {
"POSTGRES_CONNECTION_STRING": "postgresql://user:pass@localhost/db"
}
},
"custom-server": {
"command": "python",
"args": ["/path/to/custom/server.py"]
}
}
}6.2 Verify MCP Server Connection
Once configured, MCP servers automatically connect when Claude Code starts.
Available MCP tools in Claude Code:
// All MCP tools are prefixed with "mcp__<server-name>__<tool-name>"
mcp__linear__create_issue
mcp__linear__list_issues
mcp__linear__update_issue
mcp__playwright__browser_navigate
mcp__playwright__browser_screenshot
mcp__context7__resolve_library_id
mcp__context7__get_library_docsPhase 7: Security & Best Practices
7.1 Security Considerations (2025 Update)
Known Security Issues (April 2025):
- Prompt Injection: MCP tools can be exploited via prompt injection attacks
- Tool Permissions: Combining tools can enable unintended file exfiltration
- Lookalike Tools: Malicious tools can silently replace trusted ones
- Credential Exposure: API keys and credentials must be protected
Security Best Practices:
1. Validate Tool Inputs
@mcp.tool()
def read_file(filepath: str) -> str:
"""Read a file - with security validation"""
# Validate filepath to prevent path traversal
import os
filepath = os.path.normpath(filepath)
# Ensure within allowed directory
allowed_dir = "/home/user/documents"
if not filepath.startswith(allowed_dir):
raise ValueError("Access denied: Path outside allowed directory")
with open(filepath, 'r') as f:
return f.read()2. Implement Authentication
from fastmcp import FastMCP
import os
mcp = FastMCP("Secure Server", version="1.0.0")
@mcp.middleware
async def authenticate(request, call_next):
api_key = request.headers.get("X-API-Key")
expected_key = os.getenv("MCP_API_KEY")
if api_key != expected_key:
raise PermissionError("Invalid API key")
return await call_next(request)3. Rate Limiting
from collections import defaultdict
import time
call_counts = defaultdict(list)
@mcp.middleware
async def rate_limit(request, call_next):
client_id = request.client_id
now = time.time()
# Clean old entries
call_counts[client_id] = [t for t in call_counts[client_id] if now - t < 60]
# Check rate limit (10 calls per minute)
if len(call_counts[client_id]) >= 10:
raise Exception("Rate limit exceeded")
call_counts[client_id].append(now)
return await call_next(request)7.2 Error Handling Best Practices
from fastmcp import FastMCP
from typing import Optional
import logging
mcp = FastMCP("Robust Server", version="1.0.0")
logger = logging.getLogger(__name__)
@mcp.tool()
def safe_api_call(endpoint: str, params: Optional[dict] = None) -> dict:
"""Make an API call with comprehensive error handling"""
try:
# Validate inputs
if not endpoint.startswith("https://"):
raise ValueError("Only HTTPS endpoints allowed")
# Make API call
response = requests.get(endpoint, params=params, timeout=5)
response.raise_for_status()
return {
"success": True,
"data": response.json()
}
except requests.Timeout:
logger.error(f"Timeout calling {endpoint}")
return {
"success": False,
"error": "Request timeout",
"error_type": "timeout"
}
except requests.HTTPError as e:
logger.error(f"HTTP error calling {endpoint}: {e}")
return {
"success": False,
"error": str(e),
"error_type": "http_error",
"status_code": e.response.status_code
}
except Exception as e:
logger.exception(f"Unexpected error calling {endpoint}")
return {
"success": False,
"error": "Internal server error",
"error_type": "internal_error"
}7.3 Testing MCP Servers
Integration Testing (Python):
import pytest
from mcp.server import McpServer
from mcp.client import McpClient
@pytest.mark.asyncio
async def test_mcp_server_integration():
# Start test server
server = McpServer()
server.register_tool(WeatherForecastTool(MockWeatherService()))
await server.start(port=5000)
try:
# Create client
client = McpClient("http://localhost:5000")
# Test tool discovery
tools = await client.discover_tools()
assert "weatherForecast" in [t.name for t in tools]
# Test tool execution
response = await client.execute_tool("weatherForecast", {
"location": "Seattle",
"days": 3
})
# Verify response
assert response.status_code == 200
assert "Seattle" in response.content[0].text
finally:
await server.stop()Phase 8: Advanced MCP Patterns
8.1 Chain of Tools Workflow
class ChainWorkflow:
def __init__(self, tools_chain):
self.tools_chain = tools_chain
async def execute(self, mcp_client, initial_input):
current_result = initial_input
all_results = {"input": initial_input}
for tool_name in self.tools_chain:
response = await mcp_client.execute_tool(tool_name, current_result)
all_results[tool_name] = response.result
current_result = response.result
return {
"final_result": current_result,
"all_results": all_results
}
# Usage
data_pipeline = ChainWorkflow([
"dataFetch",
"dataCleaner",
"dataAnalyzer",
"dataVisualizer"
])
result = await data_pipeline.execute(
mcp_client,
{"source": "sales_database", "table": "transactions"}
)8.2 Parallel Tool Execution
async function executeParallelTools(mcpClient: Client, tools: ToolCall[]) {
const promises = tools.map(tool =>
mcpClient.callTool({
name: tool.name,
arguments: tool.arguments
})
);
const results = await Promise.all(promises);
return results;
}
// Usage
const toolCalls = [
{ name: "get_weather", arguments: { location: "Seattle" } },
{ name: "get_weather", arguments: { location: "Portland" } },
{ name: "get_weather", arguments: { location: "Vancouver" } }
];
const weatherData = await executeParallelTools(mcpClient, toolCalls);8.3 Context-Aware Tool Selection
from typing import List, Dict
import openai
class ContextAwareToolSelector:
def __init__(self, mcp_client, llm_client):
self.mcp_client = mcp_client
self.llm_client = llm_client
async def select_tools(self, user_query: str, available_tools: List[Dict]) -> List[str]:
"""Use LLM to intelligently select which tools to use"""
tool_descriptions = "\n".join([
f"- {tool['name']}: {tool['description']}"
for tool in available_tools
])
prompt = f"""
User query: {user_query}
Available tools:
{tool_descriptions}
Select the most relevant tools to answer this query.
Return a JSON array of tool names.
"""
response = await self.llm_client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
selected_tools = json.loads(response.choices[0].message.content)
return selected_tools["tools"]MCP Ecosystem (2025)
Major Platform Adoptions
OpenAI (March 2025):
- MCP integrated into ChatGPT Desktop
- MCP support in Agents SDK
- MCP compatibility in Responses API
Google (April 2025):
- MCP support in Gemini models
- Data Commons MCP Server (public datasets)
- MCP integration in Google DeepMind infrastructure
Microsoft (2025):
- MCP in Copilot Studio (GA)
- Semantic Kernel integration
- Azure OpenAI compatibility
Anthropic:
- Native MCP support in Claude Code
- Reference MCP server implementations
- MCP specification maintenance
Official MCP Servers
Repository: https://github.com/modelcontextprotocol
Enterprise Integrations:
- Google Drive, Slack, GitHub, GitLab
- Postgres, MySQL, MongoDB
- AWS, Azure, GCP
- Stripe, Salesforce
Development Tools:
- Puppeteer, Playwright
- Docker, Kubernetes
- Git, GitHub Actions
MCP SDK Support (2025)
Official SDKs:
- Python:
pip install modelcontextprotocol - TypeScript:
npm install @modelcontextprotocol/sdk - C#: NuGet package
- Java: Maven/Gradle
- Rust: Cargo package
- Swift: Swift Package Manager
Documentation Research Workflow
Using Context7 for MCP Research
When researching MCP servers and integration patterns:
# 1. Resolve library ID
/ctx7 model context protocol
# 2. Get comprehensive documentation
# Context7 returns documentation with:
# - Trust Score (7-10 for quality sources)
# - Code Snippets (100-5000+)
# - Implementation examples
# - Best practices
# 3. Evaluate quality
# - Trust Score 9-10: Official documentation
# - Trust Score 7-9: Well-maintained projects
# - Code Snippets 500+: Comprehensive coverageBest Context7 Sources for MCP (by Trust Score): 1. /microsoft/mcp-for-beginners (Trust: 9.9, Snippets: 30,945) 2. /modelcontextprotocol/python-sdk (Trust: 7.8, Snippets: 119) 3. /modelcontextprotocol/typescript-sdk (Trust: 7.8, Snippets: 55) 4. /modelcontextprotocol/csharp-sdk (Trust: 7.8, Snippets: 59)
Common MCP Integration Patterns
Pattern 1: Simple Tool Integration
Use Case: Add a single MCP tool to Claude Code
// claude_desktop_config.json
{
"mcpServers": {
"weather": {
"command": "python",
"args": ["weather_server.py"]
}
}
}# weather_server.py
from fastmcp import FastMCP
from fastmcp.transports.stdio import serve_stdio
import asyncio
mcp = FastMCP("Weather", version="1.0.0")
@mcp.tool()
def get_weather(location: str) -> str:
return f"Weather in {location}: Sunny, 72°F"
asyncio.run(serve_stdio(mcp))Pattern 2: Multi-Server Integration
Use Case: Combine multiple MCP servers for complex workflows
{
"mcpServers": {
"linear": {
"command": "npx",
"args": ["-y", "@linear/mcp-server"],
"env": { "LINEAR_API_KEY": "..." }
},
"github": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-github"],
"env": { "GITHUB_TOKEN": "..." }
},
"playwright": {
"command": "npx",
"args": ["-y", "@playwright/mcp-server"]
}
}
}Pattern 3: Custom Resource Server
Use Case: Expose internal documentation or data
from fastmcp import FastMCP
from fastmcp.transports.stdio import serve_stdio
import asyncio
import os
mcp = FastMCP("Internal Docs", version="1.0.0")
@mcp.resource("docs://{path}")
def get_documentation(path: str) -> str:
"""Get internal documentation by path"""
docs_dir = "/company/docs"
filepath = os.path.join(docs_dir, path)
if not os.path.exists(filepath):
return f"Documentation not found: {path}"
with open(filepath, 'r') as f:
return f.read()
@mcp.tool()
def search_docs(query: str) -> list:
"""Search internal documentation"""
# Implementation would use search index
return [
{"title": "Getting Started", "path": "onboarding/getting-started.md"},
{"title": "API Reference", "path": "api/reference.md"}
]
asyncio.run(serve_stdio(mcp))Troubleshooting MCP Integration
Issue 1: MCP Server Not Connecting
Symptoms: Tools not appearing in Claude Code
Solutions: 1. Check configuration file location 2. Verify command and args are correct 3. Test server independently: python server.py 4. Check logs in Claude Code 5. Ensure environment variables are set
Issue 2: Tool Execution Failures
Symptoms: Tool calls return errors
Solutions: 1. Validate tool parameter schemas 2. Add error handling in tool implementation 3. Check server logs for exceptions 4. Test with simple inputs first 5. Verify authentication/API keys
Issue 3: Performance Issues
Symptoms: Slow tool responses
Solutions: 1. Add caching for repeated calls 2. Implement connection pooling 3. Use async/await properly 4. Add timeout handling 5. Consider HTTP transport for remote servers
Best Practices Summary
1. Start Simple: Begin with a single tool, expand gradually 2. Use Official SDKs: Leverage maintained libraries (Python, TypeScript, C#, Java) 3. Implement Security: Validate inputs, authenticate, rate limit 4. Handle Errors Gracefully: Return structured error responses 5. Document Thoroughly: Provide clear tool descriptions and parameter schemas 6. Test Extensively: Write integration tests for all tools 7. Monitor Performance: Track response times and error rates 8. Version Your Servers: Use semantic versioning for changes 9. Leverage Context7: Research documentation before implementation 10. Follow MCP Specification: Adhere to official protocol standards
Quick Reference
MCP Server Checklist
- [ ] Server name and version defined
- [ ] Tools have clear descriptions
- [ ] Parameter schemas are complete
- [ ] Error handling implemented
- [ ] Authentication/authorization configured
- [ ] Logging enabled
- [ ] Tests written
- [ ] Documentation created
- [ ] Security validated
- [ ] Performance tested
Essential Commands
# Install MCP SDKs
pip install modelcontextprotocol
npm install @modelcontextprotocol/sdk
# Test server independently
python server.py
# Verify Claude Code config
cat ~/Library/Application\ Support/Claude/claude_desktop_config.json
# Research with Context7
/ctx7 [library-name]Resources
Official Documentation:
- MCP Specification: https://modelcontextprotocol.io/specification
- MCP GitHub: https://github.com/modelcontextprotocol
- MCP Servers: https://github.com/modelcontextprotocol (servers directory)
SDKs:
- Python SDK: https://github.com/modelcontextprotocol/python-sdk
- TypeScript SDK: https://github.com/modelcontextprotocol/typescript-sdk
- C# SDK: https://github.com/modelcontextprotocol/csharp-sdk
- Java SDK: https://github.com/modelcontextprotocol/java-sdk
Learning Resources:
- Microsoft MCP for Beginners: https://github.com/microsoft/mcp-for-beginners
- Anthropic MCP Announcement: https://www.anthropic.com/news/model-context-protocol
---
Skill Version: 1.0.0 Last Updated: 2025-10-18 Context7 Research: Microsoft MCP for Beginners (Trust: 9.9)
MCP Integration Examples
Comprehensive collection of practical MCP integration examples across multiple languages and use cases.
Table of Contents
1. Basic MCP Servers 2. Advanced MCP Servers 3. MCP Clients 4. LLM Integration 5. Real-World Use Cases 6. Testing 7. Security Patterns 8. Advanced Workflows
---
Basic MCP Servers
Example 1: Simple Weather Server (Python)
#!/usr/bin/env python3
"""
Simple weather MCP server demonstrating basic tool implementation.
"""
from fastmcp import FastMCP
from fastmcp.transports.stdio import serve_stdio
import asyncio
from typing import Dict
mcp = FastMCP(
name="Weather MCP Server",
version="1.0.0"
)
@mcp.tool()
def get_weather(location: str) -> Dict:
"""
Gets current weather for a location.
Args:
location: City name or coordinates
Returns:
Weather data dictionary
"""
# In production, this would call a real weather API
return {
"temperature": 72.5,
"conditions": "Sunny",
"location": location,
"humidity": 45,
"wind_speed": 8.5
}
@mcp.tool()
def get_forecast(location: str, days: int = 3) -> Dict:
"""
Gets weather forecast for multiple days.
Args:
location: City name
days: Number of days to forecast (1-7)
Returns:
Forecast data dictionary
"""
if days < 1 or days > 7:
raise ValueError("Days must be between 1 and 7")
forecast = []
for i in range(days):
forecast.append({
"day": i + 1,
"temperature": 70 + i * 2,
"conditions": "Partly Cloudy" if i % 2 == 0 else "Sunny"
})
return {
"location": location,
"forecast": forecast
}
if __name__ == "__main__":
asyncio.run(serve_stdio(mcp))Usage in Claude Code:
{
"mcpServers": {
"weather": {
"command": "python",
"args": ["/path/to/weather_server.py"]
}
}
}Example 2: Simple Calculator Server (TypeScript)
/**
* Simple calculator MCP server demonstrating TypeScript implementation.
*/
import { FastMCP } from '@modelcontextprotocol/typescript-sdk';
import { serve_stdio } from '@modelcontextprotocol/typescript-sdk/transports/stdio';
const mcp = new FastMCP({
name: "Calculator MCP Server",
version: "1.0.0"
});
mcp.tool("add", {
description: "Add two numbers",
parameters: {
a: { type: "number", required: true, description: "First number" },
b: { type: "number", required: true, description: "Second number" }
}
}, async (params) => {
return {
operation: "addition",
result: params.a + params.b,
expression: `${params.a} + ${params.b} = ${params.a + params.b}`
};
});
mcp.tool("multiply", {
description: "Multiply two numbers",
parameters: {
a: { type: "number", required: true },
b: { type: "number", required: true }
}
}, async (params) => {
return {
operation: "multiplication",
result: params.a * params.b,
expression: `${params.a} × ${params.b} = ${params.a * params.b}`
};
});
mcp.tool("calculate", {
description: "Evaluate a mathematical expression",
parameters: {
expression: { type: "string", required: true }
}
}, async (params) => {
try {
// Security: Use a safe eval method in production
const result = eval(params.expression);
return {
expression: params.expression,
result: result
};
} catch (error) {
return {
error: "Invalid expression",
message: error.message
};
}
});
serve_stdio(mcp);Example 3: File System Server (Java)
/**
* File system MCP server demonstrating Java implementation.
*/
import io.modelcontextprotocol.server.McpServer;
import io.modelcontextprotocol.server.McpToolDefinition;
import io.modelcontextprotocol.server.transport.StdioServerTransport;
import io.modelcontextprotocol.server.tool.ToolExecutionContext;
import io.modelcontextprotocol.server.tool.ToolResponse;
import java.nio.file.*;
import java.io.IOException;
import java.util.stream.Collectors;
public class FileSystemMcpServer {
private static final String SAFE_DIRECTORY = "/safe/workspace";
public static void main(String[] args) throws Exception {
McpServer server = McpServer.builder()
.name("FileSystem MCP Server")
.version("1.0.0")
.build();
// List files in directory
server.registerTool(McpToolDefinition.builder("listFiles")
.description("List files in a directory")
.parameter("path", String.class)
.execute((ToolExecutionContext ctx) -> {
String path = ctx.getParameter("path", String.class);
Path dirPath = validatePath(path);
try {
String files = Files.list(dirPath)
.map(Path::getFileName)
.map(Path::toString)
.collect(Collectors.joining("\n"));
return ToolResponse.content(files);
} catch (IOException e) {
return ToolResponse.error("Failed to list files: " + e.getMessage());
}
})
.build());
// Read file
server.registerTool(McpToolDefinition.builder("readFile")
.description("Read contents of a file")
.parameter("filepath", String.class)
.execute((ToolExecutionContext ctx) -> {
String filepath = ctx.getParameter("filepath", String.class);
Path filePath = validatePath(filepath);
try {
String content = Files.readString(filePath);
return ToolResponse.content(content);
} catch (IOException e) {
return ToolResponse.error("Failed to read file: " + e.getMessage());
}
})
.build());
try (StdioServerTransport transport = new StdioServerTransport()) {
server.connect(transport);
System.err.println("FileSystem MCP Server started");
Thread.currentThread().join();
}
}
private static Path validatePath(String path) throws SecurityException {
Path normalized = Paths.get(SAFE_DIRECTORY, path).normalize();
if (!normalized.startsWith(SAFE_DIRECTORY)) {
throw new SecurityException("Access denied: Path outside safe directory");
}
return normalized;
}
}---
Advanced MCP Servers
Example 4: Database Query Server with Connection Pool (Python)
"""
Advanced database MCP server with connection pooling and transactions.
"""
from fastmcp import FastMCP
from fastmcp.transports.stdio import serve_stdio
import asyncio
import asyncpg
from typing import Dict, List, Optional
import os
mcp = FastMCP("Database Server", version="1.0.0")
# Connection pool
pool: Optional[asyncpg.Pool] = None
async def get_pool() -> asyncpg.Pool:
global pool
if pool is None:
pool = await asyncpg.create_pool(
host=os.getenv("DB_HOST", "localhost"),
port=int(os.getenv("DB_PORT", "5432")),
user=os.getenv("DB_USER"),
password=os.getenv("DB_PASSWORD"),
database=os.getenv("DB_NAME"),
min_size=2,
max_size=10
)
return pool
@mcp.tool()
async def query(sql: str, params: Optional[List] = None) -> Dict:
"""
Execute a read-only SQL query.
Args:
sql: SELECT query to execute
params: Query parameters (optional)
Returns:
Query results as list of dictionaries
"""
# Security: Only allow SELECT
if not sql.strip().upper().startswith("SELECT"):
return {
"error": "Only SELECT queries are allowed",
"error_type": "security_violation"
}
try:
pool = await get_pool()
async with pool.acquire() as connection:
rows = await connection.fetch(sql, *(params or []))
return {
"success": True,
"rows": [dict(row) for row in rows],
"count": len(rows)
}
except Exception as e:
return {
"success": False,
"error": str(e),
"error_type": type(e).__name__
}
@mcp.tool()
async def get_table_schema(table_name: str) -> Dict:
"""Get schema information for a table."""
sql = """
SELECT column_name, data_type, is_nullable
FROM information_schema.columns
WHERE table_name = $1
ORDER BY ordinal_position
"""
try:
pool = await get_pool()
async with pool.acquire() as connection:
rows = await connection.fetch(sql, table_name)
return {
"success": True,
"table": table_name,
"columns": [dict(row) for row in rows]
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
@mcp.tool()
async def execute_transaction(queries: List[Dict]) -> Dict:
"""
Execute multiple queries in a transaction.
Args:
queries: List of {sql, params} dictionaries
Returns:
Transaction results
"""
try:
pool = await get_pool()
async with pool.acquire() as connection:
async with connection.transaction():
results = []
for query in queries:
sql = query["sql"]
params = query.get("params", [])
result = await connection.fetch(sql, *params)
results.append({
"sql": sql,
"affected_rows": len(result)
})
return {
"success": True,
"results": results,
"committed": True
}
except Exception as e:
return {
"success": False,
"error": str(e),
"rolled_back": True
}
if __name__ == "__main__":
asyncio.run(serve_stdio(mcp))Example 5: REST API Wrapper Server (TypeScript)
/**
* REST API wrapper MCP server with caching and rate limiting.
*/
import { FastMCP } from '@modelcontextprotocol/typescript-sdk';
import { serve_stdio } from '@modelcontextprotocol/typescript-sdk/transports/stdio';
import axios from 'axios';
import NodeCache from 'node-cache';
import rateLimit from 'express-rate-limit';
const mcp = new FastMCP({
name: "API Wrapper Server",
version: "1.0.0"
});
// Cache with 5-minute TTL
const cache = new NodeCache({ stdTTL: 300 });
// Rate limiter: 10 requests per minute
const rateLimiter = new Map();
function checkRateLimit(clientId: string): boolean {
const now = Date.now();
const clientRequests = rateLimiter.get(clientId) || [];
// Remove requests older than 1 minute
const recentRequests = clientRequests.filter(
(time: number) => now - time < 60000
);
if (recentRequests.length >= 10) {
return false;
}
recentRequests.push(now);
rateLimiter.set(clientId, recentRequests);
return true;
}
mcp.tool("fetchAPI", {
description: "Fetch data from a REST API with caching",
parameters: {
url: { type: "string", required: true },
method: { type: "string", required: false, default: "GET" },
headers: { type: "object", required: false },
body: { type: "object", required: false },
useCache: { type: "boolean", required: false, default: true }
}
}, async (params, context) => {
const clientId = context.clientId || "default";
// Check rate limit
if (!checkRateLimit(clientId)) {
return {
success: false,
error: "Rate limit exceeded",
retryAfter: 60
};
}
// Security: Only allow HTTPS
if (!params.url.startsWith("https://")) {
return {
success: false,
error: "Only HTTPS URLs are allowed"
};
}
const cacheKey = `${params.method}:${params.url}`;
// Check cache
if (params.useCache) {
const cached = cache.get(cacheKey);
if (cached) {
return {
success: true,
data: cached,
cached: true
};
}
}
try {
const response = await axios({
method: params.method,
url: params.url,
headers: params.headers,
data: params.body,
timeout: 5000
});
// Cache successful GET requests
if (params.method === "GET" && params.useCache) {
cache.set(cacheKey, response.data);
}
return {
success: true,
data: response.data,
status: response.status,
cached: false
};
} catch (error) {
return {
success: false,
error: error.message,
status: error.response?.status
};
}
});
serve_stdio(mcp);---
MCP Clients
Example 6: Python MCP Client with Tool Discovery
"""
Python MCP client demonstrating tool discovery and execution.
"""
from mcp.client import stdio_client, ClientSession
import asyncio
from typing import List, Dict
class MCPClientWrapper:
def __init__(self, server_command: str, server_args: List[str]):
self.server_params = {
"command": server_command,
"args": server_args
}
self.tools = []
async def connect_and_discover(self):
"""Connect to server and discover available tools."""
async with stdio_client(self.server_params) as (reader, writer):
async with ClientSession(reader, writer) as session:
await session.initialize()
# Discover tools
self.tools = await session.list_tools()
print(f"Discovered {len(self.tools)} tools:")
for tool in self.tools:
print(f" - {tool.name}: {tool.description}")
return session
async def call_tool(self, tool_name: str, arguments: Dict) -> Dict:
"""Call a tool by name with given arguments."""
async with stdio_client(self.server_params) as (reader, writer):
async with ClientSession(reader, writer) as session:
await session.initialize()
result = await session.call_tool(tool_name, arguments=arguments)
return result
async def run_workflow(self, workflow: List[Dict]):
"""
Execute a workflow of tool calls.
Args:
workflow: List of {tool, args} dictionaries
"""
results = []
async with stdio_client(self.server_params) as (reader, writer):
async with ClientSession(reader, writer) as session:
await session.initialize()
for step in workflow:
print(f"Executing: {step['tool']}")
result = await session.call_tool(
step["tool"],
arguments=step.get("args", {})
)
results.append({
"tool": step["tool"],
"result": result
})
print(f" Result: {result}")
return results
async def main():
# Create client
client = MCPClientWrapper("python", ["weather_server.py"])
# Discover tools
await client.connect_and_discover()
# Call a single tool
weather = await client.call_tool("get_weather", {"location": "Seattle"})
print(f"\nWeather: {weather}")
# Run a workflow
workflow = [
{"tool": "get_weather", "args": {"location": "Seattle"}},
{"tool": "get_weather", "args": {"location": "Portland"}},
{"tool": "get_forecast", "args": {"location": "Seattle", "days": 3}}
]
results = await client.run_workflow(workflow)
print(f"\nWorkflow results: {results}")
if __name__ == "__main__":
asyncio.run(main())Example 7: TypeScript MCP Client Class
/**
* TypeScript MCP client with comprehensive error handling.
*/
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
import { Transport } from "@modelcontextprotocol/sdk/shared/transport.js";
interface ToolCall {
name: string;
arguments: any;
}
interface ToolResult {
success: boolean;
data?: any;
error?: string;
}
class MCPClient {
private client: Client;
private transport: Transport | null = null;
private connected: boolean = false;
constructor() {
this.client = new Client({
name: "typescript-mcp-client",
version: "1.0.0"
}, {
capabilities: {
prompts: {},
resources: {},
tools: {}
}
});
}
async connect(serverCommand: string, serverArgs: string[]): Promise<void> {
this.transport = new StdioClientTransport({
command: serverCommand,
args: serverArgs
});
await this.client.connect(this.transport);
this.connected = true;
console.log("Connected to MCP server");
}
async disconnect(): Promise<void> {
if (this.transport) {
await this.transport.close();
this.connected = false;
console.log("Disconnected from MCP server");
}
}
async listTools(): Promise<any[]> {
if (!this.connected) {
throw new Error("Client not connected");
}
const tools = await this.client.listTools();
return tools;
}
async callTool(toolCall: ToolCall): Promise<ToolResult> {
if (!this.connected) {
throw new Error("Client not connected");
}
try {
const result = await this.client.callTool({
name: toolCall.name,
arguments: toolCall.arguments
});
return {
success: true,
data: result
};
} catch (error) {
return {
success: false,
error: error.message
};
}
}
async executeParallel(toolCalls: ToolCall[]): Promise<ToolResult[]> {
const promises = toolCalls.map(call => this.callTool(call));
return await Promise.all(promises);
}
async executeSequential(toolCalls: ToolCall[]): Promise<ToolResult[]> {
const results: ToolResult[] = [];
for (const call of toolCalls) {
const result = await this.callTool(call);
results.push(result);
// Stop on first error
if (!result.success) {
break;
}
}
return results;
}
}
// Usage
async function main() {
const client = new MCPClient();
try {
await client.connect("python", ["weather_server.py"]);
const tools = await client.listTools();
console.log("Available tools:", tools);
const result = await client.callTool({
name: "get_weather",
arguments: { location: "Seattle" }
});
console.log("Result:", result);
} finally {
await client.disconnect();
}
}---
LLM Integration
Example 8: OpenAI + MCP Integration (TypeScript)
/**
* Complete OpenAI + MCP integration demonstrating full workflow.
*/
import OpenAI from "openai";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
import { z } from "zod";
class OpenAIMCPIntegration {
private openai: OpenAI;
private mcpClient: Client;
private chatHistory: OpenAI.Chat.ChatCompletionMessageParam[] = [];
constructor(openaiApiKey: string) {
this.openai = new OpenAI({ apiKey: openaiApiKey });
this.mcpClient = new Client({
name: "openai-mcp-client",
version: "1.0.0"
}, {
capabilities: { tools: {} }
});
}
async connectToMCPServer(command: string, args: string[]) {
const transport = new StdioClientTransport({ command, args });
await this.mcpClient.connect(transport);
}
mcpToolToOpenAITool(tool: any): OpenAI.Chat.ChatCompletionTool {
return {
type: "function" as const,
function: {
name: tool.name,
description: tool.description,
parameters: {
type: "object",
properties: tool.input_schema.properties,
required: tool.input_schema.required || [],
},
},
};
}
async processUserMessage(userMessage: string): Promise<string> {
// Add user message to history
this.chatHistory.push({
role: "user",
content: userMessage
});
// Get MCP tools and convert to OpenAI format
const mcpTools = await this.mcpClient.listTools();
const openaiTools = mcpTools.map(t => this.mcpToolToOpenAITool(t));
// Call OpenAI
let response = await this.openai.chat.completions.create({
model: "gpt-4",
messages: this.chatHistory,
tools: openaiTools,
});
let assistantMessage = response.choices[0].message;
// Handle tool calls
while (assistantMessage.tool_calls) {
// Add assistant message to history
this.chatHistory.push(assistantMessage);
// Execute all tool calls
for (const toolCall of assistantMessage.tool_calls) {
console.log(`Calling MCP tool: ${toolCall.function.name}`);
const result = await this.mcpClient.callTool({
name: toolCall.function.name,
arguments: JSON.parse(toolCall.function.arguments),
});
// Add tool result to history
this.chatHistory.push({
role: "tool",
tool_call_id: toolCall.id,
content: JSON.stringify(result)
});
}
// Get next response from OpenAI
response = await this.openai.chat.completions.create({
model: "gpt-4",
messages: this.chatHistory,
tools: openaiTools,
});
assistantMessage = response.choices[0].message;
}
// Add final assistant message to history
this.chatHistory.push(assistantMessage);
return assistantMessage.content || "No response";
}
}
// Usage
async function main() {
const integration = new OpenAIMCPIntegration(process.env.OPENAI_API_KEY!);
await integration.connectToMCPServer("python", ["weather_server.py"]);
const response1 = await integration.processUserMessage(
"What's the weather in Seattle?"
);
console.log("Assistant:", response1);
const response2 = await integration.processUserMessage(
"What about Portland?"
);
console.log("Assistant:", response2);
}
main();Example 9: Azure OpenAI + MCP Integration (C#)
/**
* Azure OpenAI + MCP integration in C#.
*/
using Azure.AI.Inference;
using Azure;
using ModelContextProtocol.Client;
using ModelContextProtocol.Protocol.Transport;
using System.Text.Json;
public class AzureOpenAIMCPIntegration
{
private readonly ChatCompletionsClient azureClient;
private readonly IClient mcpClient;
private readonly List<ChatRequestMessage> chatHistory;
public AzureOpenAIMCPIntegration(string endpoint, string token)
{
azureClient = new ChatCompletionsClient(
new Uri(endpoint),
new AzureKeyCredential(token)
);
chatHistory = new List<ChatRequestMessage>
{
new ChatRequestSystemMessage("You are a helpful assistant with access to tools.")
};
}
public async Task ConnectToMCPServer(string command, string[] args)
{
var transport = new StdioClientTransport(new()
{
Name = "MCP Server",
Command = command,
Arguments = args
});
mcpClient = await McpClientFactory.CreateAsync(transport);
}
private ChatCompletionsToolDefinition ConvertMCPToAzureTool(
string name,
string description,
JsonElement schema)
{
var functionDefinition = new FunctionDefinition(name)
{
Description = description,
Parameters = BinaryData.FromObjectAsJson(new
{
Type = "object",
Properties = schema
})
};
return new ChatCompletionsToolDefinition(functionDefinition);
}
public async Task<string> ProcessUserMessage(string userMessage)
{
chatHistory.Add(new ChatRequestUserMessage(userMessage));
// Get MCP tools
var mcpTools = await mcpClient.ListToolsAsync();
var toolDefinitions = new List<ChatCompletionsToolDefinition>();
foreach (var tool in mcpTools)
{
JsonElement propertiesElement;
tool.JsonSchema.TryGetProperty("properties", out propertiesElement);
var toolDef = ConvertMCPToAzureTool(
tool.Name,
tool.Description,
propertiesElement
);
toolDefinitions.Add(toolDef);
}
// Call Azure OpenAI
var response = await azureClient.CompleteAsync(
chatHistory,
new ChatCompletionsOptions { Tools = toolDefinitions }
);
var choice = response.Value.Choices[0];
// Handle tool calls
while (choice.FinishReason == CompletionsFinishReason.ToolCalls)
{
chatHistory.Add(new ChatRequestAssistantMessage(choice.Message));
foreach (var toolCall in choice.Message.ToolCalls)
{
Console.WriteLine($"Calling MCP tool: {toolCall.Name}");
var args = JsonSerializer.Deserialize<Dictionary<string, object>>(
toolCall.Arguments
);
var result = await mcpClient.CallToolAsync(toolCall.Name, args);
chatHistory.Add(new ChatRequestToolMessage(
JsonSerializer.Serialize(result),
toolCall.Id
));
}
response = await azureClient.CompleteAsync(
chatHistory,
new ChatCompletionsOptions { Tools = toolDefinitions }
);
choice = response.Value.Choices[0];
}
chatHistory.Add(new ChatRequestAssistantMessage(choice.Message.Content));
return choice.Message.Content;
}
}
// Usage
var integration = new AzureOpenAIMCPIntegration(
"https://models.inference.ai.azure.com",
Environment.GetEnvironmentVariable("GITHUB_TOKEN")
);
await integration.ConnectToMCPServer("python", new[] { "weather_server.py" });
var response = await integration.ProcessUserMessage("What's the weather in Seattle?");
Console.WriteLine($"Assistant: {response}");---
Real-World Use Cases
Example 10: Internal Documentation Server
"""
Internal documentation MCP server with search and access control.
"""
from fastmcp import FastMCP
from fastmcp.transports.stdio import serve_stdio
import asyncio
import os
from pathlib import Path
from typing import List, Dict
import whoosh
from whoosh.index import create_in, open_dir
from whoosh.fields import Schema, TEXT, ID
from whoosh.qparser import QueryParser
mcp = FastMCP("Internal Docs Server", version="1.0.0")
DOCS_DIR = "/company/documentation"
INDEX_DIR = "/tmp/docs_index"
# Initialize search index
schema = Schema(
path=ID(stored=True),
title=TEXT(stored=True),
content=TEXT
)
@mcp.resource("docs://{path}")
def get_document(path: str) -> str:
"""
Retrieve a document by path.
Args:
path: Relative path to document
Returns:
Document content
"""
filepath = Path(DOCS_DIR) / path
filepath = filepath.resolve()
# Security: Ensure within DOCS_DIR
if not str(filepath).startswith(DOCS_DIR):
return "Error: Access denied"
if not filepath.exists():
return f"Error: Document not found: {path}"
with open(filepath, 'r') as f:
return f.read()
@mcp.tool()
def search_docs(query: str, limit: int = 10) -> List[Dict]:
"""
Search documentation using full-text search.
Args:
query: Search query
limit: Maximum number of results
Returns:
List of matching documents
"""
try:
ix = open_dir(INDEX_DIR)
with ix.searcher() as searcher:
query_parser = QueryParser("content", ix.schema)
q = query_parser.parse(query)
results = searcher.search(q, limit=limit)
return [
{
"path": result["path"],
"title": result["title"],
"score": result.score
}
for result in results
]
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
def list_categories() -> List[str]:
"""List all documentation categories."""
categories = []
for item in Path(DOCS_DIR).iterdir():
if item.is_dir():
categories.append(item.name)
return sorted(categories)
@mcp.tool()
def get_toc(category: str) -> Dict:
"""
Get table of contents for a category.
Args:
category: Documentation category
Returns:
Hierarchical table of contents
"""
category_path = Path(DOCS_DIR) / category
if not category_path.exists():
return {"error": f"Category not found: {category}"}
toc = {}
for filepath in category_path.rglob("*.md"):
relative = filepath.relative_to(category_path)
toc[str(relative)] = {
"path": str(filepath.relative_to(DOCS_DIR)),
"size": filepath.stat().st_size,
"modified": filepath.stat().st_mtime
}
return {"category": category, "documents": toc}
if __name__ == "__main__":
asyncio.run(serve_stdio(mcp))Example 11: CI/CD Integration Server
/**
* CI/CD integration MCP server connecting GitHub, Linear, and deployment.
*/
import { FastMCP } from '@modelcontextprotocol/typescript-sdk';
import { serve_stdio } from '@modelcontextprotocol/typescript-sdk/transports/stdio';
import { Octokit } from '@octokit/rest';
import { LinearClient } from '@linear/sdk';
import axios from 'axios';
const mcp = new FastMCP({
name: "CI/CD Integration Server",
version: "1.0.0"
});
const github = new Octokit({ auth: process.env.GITHUB_TOKEN });
const linear = new LinearClient({ apiKey: process.env.LINEAR_API_KEY });
mcp.tool("createReleasePR", {
description: "Create a release PR with Linear issue tracking",
parameters: {
owner: { type: "string", required: true },
repo: { type: "string", required: true },
version: { type: "string", required: true },
changes: { type: "array", items: { type: "string" } }
}
}, async (params) => {
try {
// 1. Create Linear issue for release
const issue = await linear.issueCreate({
teamId: process.env.LINEAR_TEAM_ID!,
title: `Release ${params.version}`,
description: `Release version ${params.version}\n\nChanges:\n${params.changes.join('\n')}`,
labels: ["release"]
});
// 2. Create GitHub PR
const pr = await github.pulls.create({
owner: params.owner,
repo: params.repo,
title: `Release ${params.version}`,
head: `release/${params.version}`,
base: "main",
body: `# Release ${params.version}\n\nLinear Issue: ${issue.issue?.url}\n\n## Changes\n${params.changes.map(c => `- ${c}`).join('\n')}`
});
// 3. Link PR to Linear issue
await linear.attachmentCreate({
issueId: issue.issue!.id,
title: `PR #${pr.data.number}`,
url: pr.data.html_url
});
return {
success: true,
pr: {
number: pr.data.number,
url: pr.data.html_url
},
linear_issue: {
id: issue.issue!.id,
url: issue.issue!.url
}
};
} catch (error) {
return {
success: false,
error: error.message
};
}
});
mcp.tool("deployToStaging", {
description: "Deploy a version to staging environment",
parameters: {
version: { type: "string", required: true },
environment: { type: "string", required: false, default: "staging" }
}
}, async (params) => {
try {
// Trigger deployment webhook
const response = await axios.post(
process.env.DEPLOY_WEBHOOK_URL!,
{
version: params.version,
environment: params.environment
},
{
headers: {
'Authorization': `Bearer ${process.env.DEPLOY_TOKEN}`
}
}
);
return {
success: true,
deployment_id: response.data.id,
status_url: response.data.status_url
};
} catch (error) {
return {
success: false,
error: error.message
};
}
});
serve_stdio(mcp);---
Testing
Example 12: MCP Server Integration Tests (Python)
"""
Comprehensive integration tests for MCP server.
"""
import pytest
import asyncio
from mcp.server import McpServer
from mcp.client import McpClient
from unittest.mock import Mock, patch
@pytest.fixture
async def test_server():
"""Start a test MCP server."""
server = McpServer()
# Register test tools
@server.tool("add")
async def add(a: int, b: int) -> int:
return a + b
@server.tool("greet")
async def greet(name: str) -> str:
return f"Hello, {name}!"
await server.start(port=5555)
yield server
await server.stop()
@pytest.fixture
async def test_client(test_server):
"""Create a test MCP client."""
client = McpClient("http://localhost:5555")
yield client
await client.close()
@pytest.mark.asyncio
async def test_tool_discovery(test_client):
"""Test tool discovery."""
tools = await test_client.discover_tools()
assert len(tools) == 2
assert "add" in [t.name for t in tools]
assert "greet" in [t.name for t in tools]
@pytest.mark.asyncio
async def test_tool_execution(test_client):
"""Test tool execution."""
result = await test_client.execute_tool("add", {"a": 5, "b": 7})
assert result.status_code == 200
assert result.content[0].text == "12"
@pytest.mark.asyncio
async def test_tool_with_invalid_params(test_client):
"""Test tool with invalid parameters."""
with pytest.raises(ValueError):
await test_client.execute_tool("add", {"a": "not a number", "b": 7})
@pytest.mark.asyncio
async def test_parallel_tool_calls(test_client):
"""Test parallel tool execution."""
tasks = [
test_client.execute_tool("add", {"a": i, "b": i * 2})
for i in range(5)
]
results = await asyncio.gather(*tasks)
assert len(results) == 5
assert all(r.status_code == 200 for r in results)
@pytest.mark.asyncio
async def test_tool_error_handling(test_client):
"""Test error handling."""
with pytest.raises(Exception) as exc_info:
await test_client.execute_tool("nonexistent_tool", {})
assert "Tool not found" in str(exc_info.value)---
Security Patterns
Example 13: Secure MCP Server with Authentication
"""
MCP server with comprehensive security measures.
"""
from fastmcp import FastMCP
from fastmcp.transports.stdio import serve_stdio
import asyncio
import os
import hmac
import hashlib
from collections import defaultdict
import time
from typing import Dict, Optional
import re
mcp = FastMCP("Secure Server", version="1.0.0")
# Security configurations
API_KEY = os.getenv("MCP_API_KEY")
ALLOWED_PATHS = ["/safe/workspace", "/public/docs"]
RATE_LIMIT = 10 # requests per minute
RATE_LIMIT_WINDOW = 60 # seconds
# Track requests for rate limiting
request_tracker: Dict[str, list] = defaultdict(list)
@mcp.middleware
async def authenticate(request, call_next):
"""Verify API key authentication."""
provided_key = request.headers.get("X-API-Key")
if not provided_key:
raise PermissionError("API key required")
# Constant-time comparison to prevent timing attacks
expected_key = API_KEY.encode()
provided_key_encoded = provided_key.encode()
if not hmac.compare_digest(expected_key, provided_key_encoded):
raise PermissionError("Invalid API key")
return await call_next(request)
@mcp.middleware
async def rate_limit(request, call_next):
"""Enforce rate limiting per client."""
client_id = request.client_id or "unknown"
now = time.time()
# Clean old requests
request_tracker[client_id] = [
req_time for req_time in request_tracker[client_id]
if now - req_time < RATE_LIMIT_WINDOW
]
# Check rate limit
if len(request_tracker[client_id]) >= RATE_LIMIT:
raise Exception(
f"Rate limit exceeded. Max {RATE_LIMIT} requests per {RATE_LIMIT_WINDOW}s"
)
request_tracker[client_id].append(now)
return await call_next(request)
def validate_path(filepath: str) -> str:
"""Validate and normalize file paths."""
import os.path
# Normalize path
normalized = os.path.normpath(filepath)
# Check if within allowed paths
allowed = any(
normalized.startswith(allowed_path)
for allowed_path in ALLOWED_PATHS
)
if not allowed:
raise SecurityError(f"Access denied: {filepath}")
return normalized
def sanitize_input(value: str) -> str:
"""Sanitize user input to prevent injection attacks."""
# Remove potentially dangerous characters
sanitized = re.sub(r'[^\w\s\-./]', '', value)
return sanitized
@mcp.tool()
async def read_file(filepath: str) -> Dict:
"""
Read a file with security validation.
Args:
filepath: Path to file
Returns:
File content or error
"""
try:
validated_path = validate_path(filepath)
with open(validated_path, 'r') as f:
content = f.read()
return {
"success": True,
"content": content,
"path": validated_path
}
except SecurityError as e:
return {
"success": False,
"error": str(e),
"error_type": "security_violation"
}
except Exception as e:
return {
"success": False,
"error": str(e),
"error_type": type(e).__name__
}
@mcp.tool()
async def execute_query(query: str) -> Dict:
"""
Execute a database query with security validation.
Args:
query: SQL query (SELECT only)
Returns:
Query results or error
"""
# Security: Only allow SELECT
sanitized_query = query.strip().upper()
if not sanitized_query.startswith("SELECT"):
return {
"success": False,
"error": "Only SELECT queries allowed",
"error_type": "security_violation"
}
# Security: Block dangerous keywords
dangerous_keywords = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "EXEC"]
if any(keyword in sanitized_query for keyword in dangerous_keywords):
return {
"success": False,
"error": "Query contains forbidden keywords",
"error_type": "security_violation"
}
try:
# Execute query (implementation depends on your database)
# results = database.execute(query)
return {
"success": True,
"results": [] # Placeholder
}
except Exception as e:
return {
"success": False,
"error": str(e),
"error_type": type(e).__name__
}
class SecurityError(Exception):
pass
if __name__ == "__main__":
asyncio.run(serve_stdio(mcp))---
Advanced Workflows
Example 14: Chain of Tools Workflow
"""
Advanced workflow pattern: Chain of tools with context passing.
"""
from typing import List, Dict, Any
from mcp.client import stdio_client, ClientSession
import asyncio
class ChainWorkflow:
def __init__(self, mcp_client):
self.client = mcp_client
async def execute(self, tools_chain: List[Dict], initial_input: Dict) -> Dict:
"""
Execute a chain of tools where output flows into next tool.
Args:
tools_chain: List of {tool, input_mapping} dictionaries
initial_input: Initial input data
Returns:
Final result with all intermediate results
"""
current_data = initial_input
all_results = {"initial_input": initial_input}
for step in tools_chain:
tool_name = step["tool"]
input_mapping = step.get("input_mapping", {})
# Map current data to tool inputs
tool_args = {}
for param, source in input_mapping.items():
if source.startswith("$"):
# Reference to previous result
tool_args[param] = current_data.get(source[1:])
else:
# Static value
tool_args[param] = source
print(f"Executing: {tool_name} with {tool_args}")
# Execute tool
result = await self.client.call_tool(tool_name, arguments=tool_args)
# Store result
all_results[tool_name] = result
# Update current data for next step
if isinstance(result, dict):
current_data.update(result)
else:
current_data["result"] = result
return {
"success": True,
"final_result": current_data,
"all_results": all_results
}
# Usage example
async def main():
server_params = {
"command": "python",
"args": ["data_processing_server.py"]
}
async with stdio_client(server_params) as (reader, writer):
async with ClientSession(reader, writer) as session:
await session.initialize()
workflow = ChainWorkflow(session)
# Define workflow
data_pipeline = [
{
"tool": "fetch_data",
"input_mapping": {
"source": "sales_database",
"table": "transactions"
}
},
{
"tool": "clean_data",
"input_mapping": {
"data": "$raw_data" # Reference previous result
}
},
{
"tool": "analyze_data",
"input_mapping": {
"data": "$cleaned_data",
"metrics": ["revenue", "count", "avg_value"]
}
},
{
"tool": "generate_report",
"input_mapping": {
"analysis": "$analysis_results",
"format": "markdown"
}
}
]
result = await workflow.execute(
data_pipeline,
{"start_date": "2025-01-01", "end_date": "2025-01-31"}
)
print("Final result:", result)
asyncio.run(main())---
Summary
This examples collection demonstrates:
- Basic Servers: Simple weather, calculator, and file system servers
- Advanced Servers: Database with connection pooling, API wrapper with caching
- Clients: Python and TypeScript clients with comprehensive features
- LLM Integration: OpenAI and Azure OpenAI integration patterns
- Real-World: Documentation server, CI/CD integration
- Testing: Comprehensive integration test suite
- Security: Authentication, rate limiting, input validation
- Workflows: Chain of tools, parallel execution
All examples follow MCP best practices and include proper error handling, security measures, and documentation.
MCP Integration Expert
Expert guidance for researching, documenting, and integrating Model Context Protocol (MCP) servers and tools.
Overview
The MCP Integration Expert skill provides comprehensive knowledge for working with the Model Context Protocol (MCP), an open standard introduced by Anthropic in November 2024 that standardizes how AI applications and Large Language Models (LLMs) integrate with external data sources, tools, and systems.
Think of MCP as a "USB-C port for AI" - providing a universal, standardized interface for connecting AI models to diverse data sources and tools.
What You'll Learn
This skill covers:
- MCP Architecture: Understanding the client-server model and protocol primitives
- Server Implementation: Building custom MCP servers in Python, TypeScript, C#, Java, and Rust
- Client Integration: Connecting MCP clients to servers and calling tools
- Claude Code Integration: Configuring and using MCP servers in Claude Code
- LLM Integration: Integrating MCP with OpenAI, Azure OpenAI, and other LLMs
- Security Best Practices: Implementing authentication, validation, and rate limiting
- Advanced Patterns: Chain of tools, parallel execution, context-aware selection
- Research Workflow: Using Context7 to research MCP documentation and examples
- Troubleshooting: Diagnosing and fixing common MCP integration issues
Quick Start
1. Install MCP SDK
Python:
pip install modelcontextprotocol
pip install fastmcp # For easier server creationTypeScript:
npm install @modelcontextprotocol/sdkC#:
dotnet add package ModelContextProtocol2. Create Your First MCP Server
Python (FastMCP):
from fastmcp import FastMCP
from fastmcp.transports.stdio import serve_stdio
import asyncio
mcp = FastMCP("My First Server", version="1.0.0")
@mcp.tool()
def greet(name: str) -> str:
"""Greet someone by name"""
return f"Hello, {name}!"
if __name__ == "__main__":
asyncio.run(serve_stdio(mcp))TypeScript:
import { FastMCP } from '@modelcontextprotocol/typescript-sdk';
import { serve_stdio } from '@modelcontextprotocol/typescript-sdk/transports/stdio';
const mcp = new FastMCP({
name: "My First Server",
version: "1.0.0"
});
mcp.tool("greet", {
description: "Greet someone by name",
parameters: {
name: { type: "string", required: true }
}
}, async (params) => {
return `Hello, ${params.name}!`;
});
serve_stdio(mcp);3. Configure in Claude Code
Add to ~/Library/Application Support/Claude/claude_desktop_config.json (macOS):
{
"mcpServers": {
"my-first-server": {
"command": "python",
"args": ["/path/to/server.py"]
}
}
}Restart Claude Code, and your tools will be available!
MCP Core Concepts
Three Core Primitives
1. Resources: Expose data sources (files, databases, APIs)
@mcp.resource("docs://{path}")
def get_doc(path: str) -> str:
return read_documentation(path)2. Tools: Enable actions and operations
@mcp.tool()
def calculate(operation: str, a: float, b: float) -> float:
return perform_calculation(operation, a, b)3. Prompts: Provide reusable prompt templates
@mcp.prompt("code-review")
def code_review_prompt(language: str, code: str):
return f"Review this {language} code:\n{code}"Client-Server Architecture
┌─────────────────┐
│ MCP Client │ (Claude Code, ChatGPT, Custom App)
│ (AI App/LLM) │
└────────┬────────┘
│ MCP Protocol (stdio, HTTP/SSE)
│
┌────────┴────────┐
│ MCP Server │ (Linear, GitHub, Custom Server)
│ (Tools/Data) │
└─────────────────┘Popular MCP Servers (2025)
Official MCP Servers (https://github.com/modelcontextprotocol):
- Linear: Project management and issue tracking
- GitHub: Repository management and automation
- Playwright: Browser automation and visual testing
- Postgres: Database queries and operations
- Google Drive: File storage and retrieval
- Slack: Team communication
- Stripe: Payment processing
- Puppeteer: Web scraping
Claude Code Built-in:
- Context7: Library documentation retrieval
- Linear: Project management (pre-configured)
- Playwright: Browser automation (pre-configured)
Real-World Use Cases
1. Development Workflow Automation
{
"mcpServers": {
"linear": { "command": "npx", "args": ["-y", "@linear/mcp-server"] },
"github": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-github"] },
"playwright": { "command": "npx", "args": ["-y", "@playwright/mcp-server"] }
}
}Workflow: "Create a Linear issue, generate code, commit to GitHub, run tests with Playwright"
2. Internal Documentation Access
Create a custom MCP server to access your company's internal documentation:
@mcp.resource("internal-docs://{path}")
def get_internal_doc(path: str) -> str:
return read_from_confluence(path)
@mcp.tool()
def search_docs(query: str) -> list:
return search_internal_knowledge_base(query)3. Database Operations
@mcp.tool()
def run_query(sql: str) -> dict:
"""Execute a read-only SQL query"""
# Validate query is SELECT only
if not sql.strip().upper().startswith("SELECT"):
raise ValueError("Only SELECT queries allowed")
return execute_query(sql)Security & Best Practices
1. Input Validation
Always validate and sanitize inputs:
@mcp.tool()
def read_file(filepath: str) -> str:
import os
filepath = os.path.normpath(filepath)
# Prevent path traversal
allowed_dir = "/safe/directory"
if not filepath.startswith(allowed_dir):
raise ValueError("Access denied")
return open(filepath).read()2. Authentication
Protect your MCP servers:
@mcp.middleware
async def authenticate(request, call_next):
api_key = request.headers.get("X-API-Key")
if api_key != os.getenv("MCP_API_KEY"):
raise PermissionError("Invalid API key")
return await call_next(request)3. Error Handling
Return structured errors:
@mcp.tool()
def safe_operation(param: str) -> dict:
try:
result = perform_operation(param)
return {"success": True, "data": result}
except Exception as e:
return {
"success": False,
"error": str(e),
"error_type": type(e).__name__
}Research Workflow with Context7
When building MCP integrations, use Context7 to research:
# 1. Find MCP documentation
/ctx7 model context protocol
# 2. Research specific SDK
/ctx7 modelcontextprotocol python-sdk
# 3. Find examples
/ctx7 mcp-for-beginnersTop Context7 Sources:
- microsoft/mcp-for-beginners (Trust: 9.9, 30K+ snippets)
- modelcontextprotocol/python-sdk (Trust: 7.8)
- modelcontextprotocol/typescript-sdk (Trust: 7.8)
Major Platform Adoptions (2025)
- OpenAI (March 2025): MCP in ChatGPT Desktop, Agents SDK
- Google (April 2025): MCP in Gemini models, Data Commons server
- Microsoft (2025): MCP in Copilot Studio, Azure OpenAI
- Anthropic: Native MCP in Claude Code
File Structure
mcp-integration-expert/
├── SKILL.md # Complete skill documentation (this file)
├── README.md # Quick start and overview
└── EXAMPLES.md # Detailed code examplesNext Steps
1. Read SKILL.md: Comprehensive guide to MCP integration 2. Try Examples: See EXAMPLES.md for practical implementations 3. Build Your Server: Start with a simple tool, expand gradually 4. Configure Claude Code: Add your server to claude_desktop_config.json 5. Research: Use /ctx7 to explore MCP documentation
Resources
Official Documentation:
- MCP Specification: https://modelcontextprotocol.io/specification
- MCP GitHub: https://github.com/modelcontextprotocol
- Anthropic MCP Announcement: https://www.anthropic.com/news/model-context-protocol
Learning:
- Microsoft MCP for Beginners: https://github.com/microsoft/mcp-for-beginners
- MCP Server Examples: https://github.com/modelcontextprotocol (servers directory)
SDKs:
- Python: https://github.com/modelcontextprotocol/python-sdk
- TypeScript: https://github.com/modelcontextprotocol/typescript-sdk
- C#: https://github.com/modelcontextprotocol/csharp-sdk
- Java: https://github.com/modelcontextprotocol/java-sdk
Support
For issues or questions:
- Check SKILL.md troubleshooting section
- Review EXAMPLES.md for reference implementations
- Consult official MCP documentation
- Use Context7 for latest research:
/ctx7 model context protocol
---
Skill Version: 1.0.0 Last Updated: 2025-10-18 Maintained By: MCP Integration Expert Skill