Build Model Context Protocol servers for Claude Code integration
Provides a complete template for building MCP servers with standardized structure, error handling, and testing patterns. Use when creating new MCP servers to ensure consistency and reliability across integrations.
/plugin marketplace add mindmorass/reflex/plugin install reflex@mindmorass-reflexThis skill inherits all available tools. When active, it can use any tool Claude has access to.
Build new MCP servers following established patterns for consistency and reliability.
This skill provides a template and guidelines for building new MCP servers that integrate with the agentic workspace. All servers follow the same patterns for:
pip install mcp>=1.0.0
mcp/servers/{server-name}/
├── server.py # Main server implementation
├── requirements.txt # Python dependencies
├── test_{name}.py # Test suite
├── README.md # Server documentation
└── config.example.env # Example configuration
File: mcp/servers/{server-name}/server.py
#!/usr/bin/env python3
"""
{Server Name} MCP Server - {Brief description}.
"""
import asyncio
import json
import logging
import os
from typing import Optional
from mcp.server import Server
from mcp.server.stdio import stdio_server
# =============================================================================
# Configuration
# =============================================================================
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
# Add server-specific config here
# EXAMPLE_API_KEY = os.getenv("EXAMPLE_API_KEY")
# EXAMPLE_TIMEOUT = int(os.getenv("EXAMPLE_TIMEOUT", "30"))
# Setup logging
logging.basicConfig(
level=getattr(logging, LOG_LEVEL),
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
# =============================================================================
# Server Implementation
# =============================================================================
class {ServerName}Server:
"""
{Description of what this server does}.
Tools provided:
- tool_one: Description
- tool_two: Description
"""
def __init__(self):
self.server = Server("{server-name}")
self._validate_config()
self._setup_tools()
logger.info("{ServerName} server initialized")
def _validate_config(self):
"""Validate required configuration."""
# Example validation:
# if not EXAMPLE_API_KEY:
# raise ValueError("EXAMPLE_API_KEY environment variable required")
pass
def _setup_tools(self):
"""Register MCP tools."""
@self.server.tool()
async def example_tool(
param1: str,
param2: Optional[int] = 10
) -> str:
"""
Brief description of what this tool does.
Args:
param1: Description of param1
param2: Description of param2 (default: 10)
Returns:
JSON string with result
"""
try:
logger.debug(f"example_tool called: param1={param1}, param2={param2}")
# Implementation here
result = {
"status": "success",
"param1": param1,
"param2": param2
}
return json.dumps(result)
except Exception as e:
logger.error(f"example_tool failed: {e}")
return json.dumps({
"status": "error",
"error": str(e)
})
@self.server.tool()
async def another_tool(query: str) -> str:
"""
Another tool description.
Args:
query: The query to process
"""
try:
# Implementation
return json.dumps({"result": query})
except Exception as e:
logger.error(f"another_tool failed: {e}")
return json.dumps({"status": "error", "error": str(e)})
async def run(self):
"""Run the MCP server."""
logger.info("Starting {server-name} server...")
async with stdio_server() as (read_stream, write_stream):
await self.server.run(read_stream, write_stream)
# =============================================================================
# Entry Point
# =============================================================================
def main():
server = {ServerName}Server()
asyncio.run(server.run())
if __name__ == "__main__":
main()
File: mcp/servers/{server-name}/requirements.txt
mcp>=1.0.0
# Add server-specific dependencies below
# requests>=2.28.0
# aiohttp>=3.8.0
File: mcp/servers/{server-name}/test_{name}.py
#!/usr/bin/env python3
"""Tests for {server-name} MCP server."""
import json
import os
import sys
import pytest
sys.path.insert(0, os.path.dirname(__file__))
from server import {ServerName}Server
class Test{ServerName}Server:
"""Test suite for {ServerName}Server."""
@pytest.fixture
def server(self):
"""Create server instance for testing."""
return {ServerName}Server()
def test_server_initialization(self, server):
"""Test server initializes correctly."""
assert server.server is not None
assert server.server.name == "{server-name}"
def test_config_validation(self):
"""Test configuration validation."""
# Test with missing required config
# with pytest.raises(ValueError):
# os.environ.pop("REQUIRED_VAR", None)
# {ServerName}Server()
pass
@pytest.mark.asyncio
async def test_example_tool(self, server):
"""Test example_tool function."""
# Get the tool function
tools = server.server._tools
example_tool = tools.get("example_tool")
# Call it
result = await example_tool("test_value", 20)
data = json.loads(result)
assert data["status"] == "success"
assert data["param1"] == "test_value"
assert data["param2"] == 20
@pytest.mark.asyncio
async def test_example_tool_defaults(self, server):
"""Test example_tool with default values."""
tools = server.server._tools
example_tool = tools.get("example_tool")
result = await example_tool("test")
data = json.loads(result)
assert data["param2"] == 10 # default value
@pytest.mark.asyncio
async def test_error_handling(self, server):
"""Test error handling returns proper format."""
# Trigger an error condition and verify response format
pass
def test_imports():
"""Test all imports work."""
from server import {ServerName}Server, main
print("✅ Imports working")
def test_quick():
"""Quick smoke test."""
server = {ServerName}Server()
assert server is not None
print("✅ Quick test passed")
if __name__ == "__main__":
test_imports()
test_quick()
print("
✅ All basic tests passed!")
print("Run 'pytest test_{name}.py -v' for full test suite")
File: mcp/servers/{server-name}/README.md
# {Server Name} MCP Server
{Brief description of what this server does and why it exists.}
## Tools
| Tool | Description |
|------|-------------|
| `example_tool` | Does X with Y |
| `another_tool` | Does A with B |
## Configuration
| Variable | Required | Default | Description |
|----------|----------|---------|-------------|
| `LOG_LEVEL` | No | `INFO` | Logging level |
| `EXAMPLE_API_KEY` | Yes | - | API key for service |
## Installation
```bash
cd mcp/servers/{server-name}
pip install -r requirements.txt
Add to .claude.json:
{
"mcpServers": {
"{server-name}": {
"command": "python",
"args": ["mcp/servers/{server-name}/server.py"],
"env": {
"EXAMPLE_API_KEY": "${EXAMPLE_API_KEY}"
}
}
}
}
# Example usage of example_tool
result = await example_tool(
param1="value",
param2=42
)
# Example usage of another_tool
result = await another_tool(query="search term")
# Quick test
python test_{name}.py
# Full test suite
pytest test_{name}.py -v
{Any development notes, contribution guidelines, or known limitations.}
### Step 6: Example Config
**File: `mcp/servers/{server-name}/config.example.env`**
```bash
# {Server Name} Configuration
# Copy to .env and fill in values
# Logging
LOG_LEVEL=INFO
# Required
# EXAMPLE_API_KEY=your-api-key-here
# Optional
# EXAMPLE_TIMEOUT=30
Always return JSON with consistent structure:
# Success
{"status": "success", "data": {...}}
# Error
{"status": "error", "error": "Human-readable message"}
Use structured logging:
logger.debug(f"Tool called: {params}") # Detailed debugging
logger.info(f"Operation completed: {id}") # Normal operations
logger.warning(f"Retrying after: {error}") # Recoverable issues
logger.error(f"Operation failed: {error}") # Failures
_validate_config()After building, add to .claude.json:
{
"mcpServers": {
"{server-name}": {
"command": "python",
"args": ["mcp/servers/{server-name}/server.py"],
"env": {
"LOG_LEVEL": "INFO"
}
}
}
}
import aiohttp
class APIClient:
def __init__(self, base_url: str, api_key: str):
self.base_url = base_url
self.headers = {"Authorization": f"Bearer {api_key}"}
async def get(self, endpoint: str) -> dict:
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self.base_url}/{endpoint}",
headers=self.headers
) as resp:
return await resp.json()
import asyncpg
class Database:
def __init__(self, dsn: str):
self.dsn = dsn
self.pool = None
async def connect(self):
self.pool = await asyncpg.create_pool(self.dsn)
async def query(self, sql: str, *args):
async with self.pool.acquire() as conn:
return await conn.fetch(sql, *args)
from functools import lru_cache
from datetime import datetime, timedelta
class TTLCache:
def __init__(self, ttl_seconds: int = 300):
self.ttl = timedelta(seconds=ttl_seconds)
self.cache = {}
def get(self, key: str):
if key in self.cache:
value, expires = self.cache[key]
if datetime.now() < expires:
return value
del self.cache[key]
return None
def set(self, key: str, value):
self.cache[key] = (value, datetime.now() + self.ttl)
Add notes here as you build servers and discover improvements.
This skill should be used when the user asks to "create a slash command", "add a command", "write a custom command", "define command arguments", "use command frontmatter", "organize commands", "create command with file references", "interactive command", "use AskUserQuestion in command", or needs guidance on slash command structure, YAML frontmatter fields, dynamic arguments, bash execution in commands, user interaction patterns, or command development best practices for Claude Code.
This skill should be used when the user asks to "create an agent", "add an agent", "write a subagent", "agent frontmatter", "when to use description", "agent examples", "agent tools", "agent colors", "autonomous agent", or needs guidance on agent structure, system prompts, triggering conditions, or agent development best practices for Claude Code plugins.
This skill should be used when the user asks to "create a hook", "add a PreToolUse/PostToolUse/Stop hook", "validate tool use", "implement prompt-based hooks", "use ${CLAUDE_PLUGIN_ROOT}", "set up event-driven automation", "block dangerous commands", or mentions hook events (PreToolUse, PostToolUse, Stop, SubagentStop, SessionStart, SessionEnd, UserPromptSubmit, PreCompact, Notification). Provides comprehensive guidance for creating and implementing Claude Code plugin hooks with focus on advanced prompt-based hooks API.