Workflow-005: MCP Integration
FreshSource: Model Context Protocol Intro & MCP Advanced Topics
Document Control
| Field | Value |
|---|---|
| Workflow ID | WF-005 |
| Version | 1.0 |
| Status | Active |
| Last Updated | 2024-12 |
Overview
Integrate MCP servers and clients for extensible AI applications.
MCP Architecture
Integration Workflow
Server Implementation
python
from mcp.server import Server
from mcp.server.stdio import stdio_server
# Create server
server = Server("my-integration")
# Add tools
@server.list_tools()
async def list_tools():
return [
{
"name": "query_data",
"description": "Query the data source",
"inputSchema": {
"type": "object",
"properties": {
"query": {"type": "string"}
},
"required": ["query"]
}
}
]
@server.call_tool()
async def call_tool(name, arguments):
if name == "query_data":
result = await execute_query(arguments["query"])
return {"result": result}
# Add resources
@server.list_resources()
async def list_resources():
return [
{
"uri": "data://metrics",
"name": "Current Metrics",
"mimeType": "application/json"
}
]
@server.read_resource()
async def read_resource(uri):
if uri == "data://metrics":
return get_current_metrics()Client Configuration
Claude Desktop (claude_desktop_config.json):
json
{
"mcpServers": {
"my-integration": {
"command": "python",
"args": ["path/to/server.py"],
"env": {
"API_KEY": "your-key"
}
}
}
}Custom Client:
python
from mcp.client import Client
from mcp.client.stdio import stdio_client
async def connect_to_server():
async with stdio_client("python", ["server.py"]) as (read, write):
async with Client("my-client", read, write) as client:
# Initialize
await client.initialize()
# List available tools
tools = await client.list_tools()
# Call a tool
result = await client.call_tool(
"query_data",
{"query": "SELECT * FROM users"}
)
return resultMCP Primitives
| Primitive | Control | Description | Examples |
|---|---|---|---|
| Tools | Model-controlled | Actions with input/output schema | query_database, send_email |
| Resources | Application-controlled | URI-based data access | file://docs, db://users |
| Prompts | User-controlled | Reusable prompt templates | analyze_code, summarize_doc |
| Sampling | Server-initiated | LLM requests with human approval | Enables agentic behaviors |
Testing Strategy
python
import pytest
@pytest.mark.asyncio
async def test_server_tools():
# Connect to server
async with create_test_client() as client:
# Test tool listing
tools = await client.list_tools()
assert len(tools) > 0
# Test tool execution
result = await client.call_tool(
"query_data",
{"query": "test"}
)
assert "result" in result
@pytest.mark.asyncio
async def test_server_resources():
async with create_test_client() as client:
# Test resource listing
resources = await client.list_resources()
assert any(r["uri"] == "data://metrics" for r in resources)
# Test resource reading
content = await client.read_resource("data://metrics")
assert content is not NoneVerification Checklist
- [ ] Server starts without errors
- [ ] Tools listed correctly
- [ ] Tool execution works
- [ ] Resources accessible
- [ ] Client connects successfully
- [ ] Error handling implemented
- [ ] Logging configured