mcpify_server.py•22 kB
#!/usr/bin/env python3
"""
MCPify - The MCP Development Knowledge Base
Turns you into an MCP development expert by providing instant access to
the best examples, patterns, and documentation.
"""
import os
import json
import subprocess
import tempfile
import traceback
from pathlib import Path
from typing import Dict, Optional
from fastmcp import FastMCP
import logging
# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Initialize FastMCP
mcp = FastMCP("mcpify", version="0.1.0")
mcp.description = "MCP Development Knowledge Base - Learn to build production MCPs"
# Cache directory for repos
CACHE_DIR = Path(tempfile.gettempdir()) / "mcpify_cache"
CACHE_DIR.mkdir(exist_ok=True)
def clone_or_update_repo(repo_url: str, repo_name: str) -> Path:
"""Clone or update a repository to cache"""
repo_path = CACHE_DIR / repo_name
try:
if repo_path.exists():
# Update existing repo
logger.info(f"Updating {repo_name}...")
subprocess.run(["git", "pull"], cwd=repo_path, check=True, capture_output=True)
else:
# Clone new repo
logger.info(f"Cloning {repo_name}...")
subprocess.run(["git", "clone", repo_url, str(repo_path)], check=True, capture_output=True)
return repo_path
except subprocess.CalledProcessError as e:
logger.error(f"Failed to clone/update {repo_name}: {e}")
raise
@mcp.tool()
def get_latest_mcp_knowledge() -> str:
"""
Pulls latest MCP repos and provides guided learning path.
This is your starting point for MCP mastery!
"""
try:
# Pull the repos
mcp_repo = clone_or_update_repo(
"https://github.com/modelcontextprotocol/python-sdk.git",
"mcp-python-sdk"
)
fastmcp_repo = clone_or_update_repo(
"https://github.com/jlowin/fastmcp.git",
"fastmcp"
)
return f"""📚 **FRESH MCP KNOWLEDGE PULLED!**
🎯 **START HERE - Core Examples:**
`{mcp_repo}/src/mcp/server/` - See the actual server implementation
`{fastmcp_repo}/examples/` - Best practical examples!
📖 **Essential Reading Order:**
1. **FastMCP Examples**: `{fastmcp_repo}/examples/` (BEST starting point!)
2. **FastMCP README**: `{fastmcp_repo}/README.md`
3. **MCP Python SDK**: `{mcp_repo}/src/mcp/`
4. **Type Definitions**: `{mcp_repo}/src/mcp/types.py`
🔧 **Key Implementation Examples:**
- **Weather Tool**: `{fastmcp_repo}/examples/weather.py`
- **Filesystem**: `{fastmcp_repo}/examples/filesystem.py`
- **Git Tools**: Look for git examples
- **Simple Server**: `{mcp_repo}/src/mcp/server/stdio.py`
💡 **Advanced Patterns:**
- **Error Handling**: See weather example for proper error patterns
- **Async Operations**: Check filesystem for async file operations
- **Resource Subscriptions**: SQLite example shows dynamic resources
🚀 **Quick Start Commands:**
```bash
# Install MCP SDK
pip install mcp
# Install FastMCP (easier API)
pip install fastmcp
# Run example server
python {mcp_repo}/examples/everything/server.py
```
📝 **Next Steps:**
1. Read the everything example to understand all features
2. Pick the pattern closest to your use case
3. Use FastMCP for simpler implementation
4. Test with MCP Inspector or Claude Desktop
"""
except Exception as e:
logger.error(f"Error fetching repos: {traceback.format_exc()}")
return f"❌ Error fetching repos: {str(e)}\n\nMake sure git is installed and you have internet access."
@mcp.tool()
def mcp_development_checklist() -> str:
"""
Complete checklist for building production-ready MCPs.
Follow this to ensure your MCP is robust and user-friendly.
"""
return """✅ **MCP PRODUCTION CHECKLIST**
📋 **Core Requirements:**
□ Clear tool/resource naming (use snake_case)
□ Comprehensive docstrings for all tools
□ Input validation with Pydantic models
□ Proper error handling (don't let exceptions leak)
□ Logging for debugging (use Python logging)
🏗️ **Architecture Decisions:**
□ Choose: FastMCP (simple) vs MCP SDK (full control)
□ Choose: Tools (actions) vs Resources (data) vs Both
□ Choose: Stdio (subprocess) vs HTTP (server) transport
□ Choose: Sync vs Async operations
📊 **Status & Progress:**
□ Implement progress callbacks for long operations
□ Use status line updates for real-time feedback
□ Return structured responses (not just strings)
□ Include metadata in responses when helpful
🧪 **Testing Strategy:**
□ Unit tests for core logic
□ Integration tests with MCP Inspector
□ Test with multiple clients (Claude, Continue, etc.)
□ Test error cases and edge conditions
□ Validate with different transports
📦 **Packaging & Distribution:**
□ Create pyproject.toml or setup.py
□ Include comprehensive README
□ Document configuration options
□ Provide example .claude.json config
□ Consider Docker containerization
🔒 **Security Considerations:**
□ Validate all inputs thoroughly
□ Sanitize file paths (prevent traversal)
□ Don't expose sensitive data in errors
□ Use environment variables for secrets
□ Implement rate limiting if needed
📚 **Documentation:**
□ README with clear installation steps
□ Tool usage examples
□ Configuration documentation
□ Troubleshooting guide
□ API reference (if complex)
🚀 **Advanced Features:**
□ Implement tool confirmation for dangerous operations
□ Add resource templates for common queries
□ Support multiple output formats
□ Implement caching for expensive operations
□ Add telemetry/analytics (optional)
"""
@mcp.tool()
def common_mcp_mistakes() -> str:
"""
Learn from common MCP development mistakes.
Avoid these pitfalls to save hours of debugging!
"""
return """⚠️ **COMMON MCP MISTAKES TO AVOID**
🐛 **The Silent Failures:**
- **Not using sys.exit()**: MCPs must exit cleanly
- **Swallowing exceptions**: Always log errors before handling
- **Missing error context**: Include relevant data in error messages
🔧 **Tool Design Issues:**
- **Overly complex tools**: Break into smaller, focused tools
- **Poor naming**: "process_data" vs "convert_csv_to_json"
- **Missing validation**: Always validate inputs with Pydantic
- **Huge responses**: Paginate or limit large outputs
📡 **Transport Problems:**
- **Wrong transport type**: stdio for subprocess, HTTP for servers
- **Not handling connection drops**: Implement reconnection logic
- **Blocking operations**: Use async for long-running tasks
🎯 **Integration Gotchas:**
- **Hardcoded paths**: Use env vars or config files
- **Platform-specific code**: Test on Windows/Mac/Linux
- **Missing dependencies**: Document all requirements
- **Version conflicts**: Pin your dependencies
💾 **State Management:**
- **Global state in tools**: Tools should be stateless
- **File system assumptions**: Don't assume directories exist
- **Resource leaks**: Clean up temp files and connections
🔍 **Debugging Nightmares:**
- **No logging**: Add comprehensive logging from day 1
- **Cryptic errors**: Provide actionable error messages
- **No inspection support**: Test with MCP Inspector
📊 **Performance Issues:**
- **No caching**: Cache expensive operations
- **Synchronous everything**: Use async for I/O operations
- **Loading everything**: Implement pagination
- **No progress feedback**: Users think it's frozen
🚫 **Security Mistakes:**
- **Path traversal vulnerabilities**: Validate file paths
- **Command injection**: Never use shell=True with user input
- **Exposed secrets**: Use environment variables
- **No input sanitization**: Validate everything
💡 **Pro Tips:**
- Start with FastMCP for simpler implementation
- Test with MCP Inspector before client integration
- Use structured logging (JSON format)
- Return structured data, not just strings
- Keep tools focused and composable
"""
@mcp.tool()
def mcp_architecture_patterns() -> str:
"""
Advanced MCP architecture patterns and when to use them.
Level up from basic MCPs to production systems!
"""
return """🏗️ **MCP ARCHITECTURE PATTERNS**
📐 **Pattern 1: Simple Tool Provider**
```python
# When: Basic operations, no state needed
@mcp.tool()
def calculate_something(input: str) -> str:
return process(input)
```
Use for: Calculators, converters, stateless operations
📊 **Pattern 2: Resource Manager**
```python
# When: Exposing data, not actions
@mcp.resource("data://items/{id}")
async def get_item(id: str) -> dict:
return fetch_from_db(id)
```
Use for: Databases, APIs, file systems, data stores
🔄 **Pattern 3: Hybrid Tool+Resource**
```python
# When: Need both actions and data access
@mcp.tool()
def create_item(data: dict) -> str:
# Action that modifies state
@mcp.resource("data://items")
def list_items() -> list:
# Query current state
```
Use for: CRUD systems, state managers, complex applications
🎭 **Pattern 4: Orchestrator MCP**
```python
# When: Coordinating multiple MCPs
@mcp.tool()
def run_workflow(steps: list) -> str:
# Calls other MCPs in sequence
for step in steps:
result = call_mcp(step.mcp, step.tool, step.args)
```
Use for: Workflow automation, meta-MCPs, complex pipelines
🔌 **Pattern 5: Adapter MCP**
```python
# When: Wrapping external APIs/services
class ServiceAdapter:
def __init__(self):
self.client = ExternalAPI()
@mcp.tool()
def api_operation(self, params: dict) -> dict:
return self.client.call(params)
```
Use for: API wrappers, legacy system integration, service adapters
📡 **Pattern 6: Event Stream MCP**
```python
# When: Real-time updates needed
@mcp.resource("events://stream")
async def event_stream():
async for event in watch_events():
yield event
```
Use for: Monitoring, logs, real-time data, notifications
🧩 **Pattern 7: Plugin System MCP**
```python
# When: Extensible functionality needed
class PluginMCP:
def load_plugins(self):
for plugin in discover_plugins():
self.register_tool(plugin)
```
Use for: Extensible systems, modular architectures
🔐 **Pattern 8: Gateway MCP**
```python
# When: Need auth, rate limiting, routing
@mcp.tool()
def authenticated_operation(token: str, params: dict):
if not validate_token(token):
raise AuthError()
return forward_to_backend(params)
```
Use for: Security layers, API gateways, access control
💡 **Choosing Patterns:**
- Start simple, evolve as needed
- One pattern per MCP (usually)
- Compose MCPs for complex systems
- Consider maintenance overhead
- Think about testing strategy
"""
@mcp.tool()
def mcp_testing_guide() -> str:
"""
Comprehensive guide to testing MCPs.
Ensure your MCP works reliably across all scenarios!
"""
return """🧪 **MCP TESTING GUIDE**
📋 **Testing Layers:**
**1️⃣ Unit Tests - Test Core Logic**
```python
# test_my_mcp.py
def test_tool_logic():
# Test without MCP framework
result = my_tool_function("input")
assert result == "expected"
```
**2️⃣ Integration Tests - Test MCP Server**
```python
# Test with MCP framework
async def test_mcp_tool():
server = MyMCPServer()
result = await server.call_tool("my_tool", {"arg": "value"})
assert result.success
```
**3️⃣ Client Tests - Test with Inspector**
```bash
# Using MCP Inspector
npx @modelcontextprotocol/inspector@latest \
python path/to/your/mcp.py
# Test each tool manually
# Verify responses match expectations
```
**4️⃣ End-to-End Tests - Test with Real Clients**
```json
// .claude.json configuration test
{
"mcpServers": {
"test-mcp": {
"command": "python",
"args": ["path/to/mcp.py"]
}
}
}
```
🎯 **What to Test:**
□ **Happy Path:**
- Normal inputs → Expected outputs
- All tools/resources accessible
- Proper response formats
□ **Edge Cases:**
- Empty inputs
- Maximum size inputs
- Special characters
- Boundary values
□ **Error Cases:**
- Invalid inputs
- Missing required parameters
- Network failures
- File system errors
□ **Performance:**
- Response times
- Memory usage
- Concurrent requests
- Rate limiting
🔧 **Testing Tools:**
**MCP Inspector** - Interactive testing
```bash
npx @modelcontextprotocol/inspector@latest
```
**Pytest** - Python testing
```python
pytest tests/ -v --cov=my_mcp
```
**Locust** - Load testing
```python
from locust import HttpUser, task
class MCPUser(HttpUser):
@task
def test_tool(self):
self.client.post("/call", json={...})
```
📊 **Test Data Strategies:**
1. **Fixtures**: Predefined test data
2. **Mocks**: Simulate external dependencies
3. **Factories**: Generate test data dynamically
4. **Snapshots**: Compare against known good outputs
⚡ **Continuous Testing:**
```yaml
# .github/workflows/test.yml
name: Test MCP
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- run: pip install -r requirements.txt
- run: pytest tests/
```
🐛 **Debugging Tips:**
1. **Enable verbose logging**
```python
logging.basicConfig(level=logging.DEBUG)
```
2. **Use MCP Inspector's debug mode**
3. **Add print statements in tools**
4. **Check transport layer issues**
5. **Validate JSON serialization**
✅ **Testing Checklist:**
□ All tools have unit tests
□ Error cases are tested
□ Performance is acceptable
□ Works with MCP Inspector
□ Works with target client(s)
□ CI/CD pipeline runs tests
□ Documentation includes test examples
"""
@mcp.tool()
def how_to_nest_mcps() -> str:
"""
Advanced guide on nesting MCPs for complex architectures.
Based on real experience with OMNISANC and MCPSquared!
"""
return """🪆 **NESTING MCPs - ADVANCED PATTERN**
🎯 **Why Nest MCPs?**
- Compose complex functionality from simple MCPs
- Create meta-MCPs that orchestrate others
- Build hierarchical architectures
- Enable dynamic MCP loading
📐 **Basic Nesting Pattern:**
```python
# Parent MCP that calls child MCPs
import subprocess
import json
@mcp.tool()
def call_nested_mcp(mcp_name: str, tool_name: str, args: dict) -> dict:
\"\"\"Call a tool in a nested MCP\"\"\"
# Launch child MCP as subprocess
result = subprocess.run(
["python", "-m", mcp_name],
input=json.dumps({
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": args
}
}),
capture_output=True,
text=True
)
return json.loads(result.stdout)
```
🏗️ **Advanced Nesting - MCPSquared Pattern:**
```python
# Meta-MCP that analyzes and generates MCPs
class MCPSquared:
def __init__(self):
self.child_mcps = {}
@mcp.tool()
def analyze_mcp(self, mcp_config: dict) -> dict:
\"\"\"Analyze an MCP and generate workflows\"\"\"
# Launch MCP with config
child = self.launch_mcp(mcp_config)
# Introspect available tools
tools = child.list_tools()
# Generate workflow suggestions
workflows = self.generate_workflows(tools)
return {
"tools": tools,
"workflows": workflows
}
def launch_mcp(self, config: dict):
# Start MCP server with config
# Return client connection
pass
```
⚠️ **Critical Nesting Challenges:**
**1. Path Resolution Issues:**
```python
# PROBLEM: Nested MCPs can't find modules
# SOLUTION: Use proper package structure
# BAD: path manipulation # NEVER DO THIS!
# GOOD: pip install packages properly
```
**2. State Isolation:**
```python
# Each MCP needs isolated state
MCPS_STATE_DIR = Path(os.environ.get('MCP_STATE_DIR', '/tmp/mcp_state'))
state_file = MCPS_STATE_DIR / f"{mcp_instance_id}.json"
```
**3. Environment Variables:**
```python
# Pass env vars to nested MCPs
env = os.environ.copy()
env['NESTED_MCP_CONFIG'] = json.dumps(config)
subprocess.run(cmd, env=env)
```
**4. Error Propagation:**
```python
try:
result = call_nested_mcp(...)
except MCPError as e:
# Wrap and re-raise with context
raise ParentMCPError(f"Nested MCP {mcp_name} failed: {e}")
```
🔥 **Real-World Example - OMNISANC + MCPSquared:**
```python
# OMNISANC validates sequences across nested MCPs
class OMNISANC:
@mcp.tool()
def validate_nested_sequence(self, mcp_chain: list) -> dict:
\"\"\"Validate tool sequence across multiple MCPs\"\"\"
for mcp_config in mcp_chain:
# Launch nested MCP
mcp_instance = self.launch_mcp(mcp_config)
# Validate its tools against rules
validation = self.validate_tools(mcp_instance)
if not validation.passed:
return {"blocked": True, "reason": validation.error}
return {"allowed": True}
```
💡 **Best Practices:**
1. **Use Package Management**: Never use path manipulation
2. **Isolate State**: Each MCP gets its own state directory
3. **Handle Timeouts**: Nested calls can be slow
4. **Log Everything**: Debugging nested MCPs is hard
5. **Test Incrementally**: Test each layer separately
6. **Document Dependencies**: Clear requirements.txt
7. **Use Type Hints**: Makes nesting clearer
🚀 **When to Use Nesting:**
✅ Meta-analysis tools (MCPSquared)
✅ Workflow orchestration (OMNISANC)
✅ Dynamic MCP loading
✅ Complex architectures
❌ **When NOT to Nest:**
- Simple tools can be in one MCP
- Performance critical paths
- When debugging is difficult
- Circular dependencies possible
"""
@mcp.tool()
def sdk_for_ui_over_mcp() -> str:
"""
Get the MCP-UI SDK for building user interfaces over MCP servers.
This repo shows how to create web UIs that interact with MCP servers.
"""
try:
# Pull the MCP-UI repo
mcp_ui_repo = clone_or_update_repo(
"https://github.com/idosal/mcp-ui.git",
"mcp-ui"
)
return f"""🎨 **MCP-UI SDK PULLED!**
🖥️ **UI Framework for MCP:**
`{mcp_ui_repo}/` - Complete MCP-UI implementation
📖 **Key Files to Study:**
- `{mcp_ui_repo}/README.md` - Setup and usage instructions
- `{mcp_ui_repo}/src/` - Core UI components and MCP integration
- `{mcp_ui_repo}/examples/` - Example implementations (if available)
- `{mcp_ui_repo}/package.json` - Dependencies and scripts
🎯 **What This Provides:**
- Web-based interface for MCP servers
- Real-time interaction with MCP tools and resources
- Modern UI framework integration with MCP protocol
- Example patterns for building custom MCP UIs
🚀 **Quick Start:**
```bash
cd {mcp_ui_repo}
npm install
npm start
```
💡 **Use Cases:**
- Dashboard for MCP server management
- Interactive tool testing interface
- Custom business applications over MCP
- Debugging and development tools
📝 **Next Steps:**
1. Explore the source code structure
2. Run the example implementation
3. Adapt patterns for your use case
4. Build custom UI components for your MCP tools
"""
except Exception as e:
logger.error(f"Error fetching MCP-UI repo: {traceback.format_exc()}")
return f"❌ Error fetching MCP-UI repo: {str(e)}\n\nMake sure git is installed and you have internet access."
@mcp.tool()
def payments_over_cloudflare_using_mcp() -> str:
"""
Get payment processing examples using MCP over Cloudflare Workers.
Shows how to build paid MCP services with authentication and billing.
"""
try:
# Pull both relevant repos
mcp_boilerplate = clone_or_update_repo(
"https://github.com/iannuttall/mcp-boilerplate.git",
"mcp-boilerplate"
)
agent_mcp = clone_or_update_repo(
"https://github.com/rinadelph/Agent-MCP.git",
"agent-mcp"
)
return f"""💳 **PAYMENT MCP EXAMPLES PULLED!**
🏗️ **MCP Boilerplate (Production-Ready Payments):**
`{mcp_boilerplate}/` - Complete paid MCP server template
📖 **Key Payment Files:**
- `{mcp_boilerplate}/src/tools/` - Paid tool implementations
- `{mcp_boilerplate}/src/auth/` - Authentication handlers
- `{mcp_boilerplate}/.dev.vars.example` - Environment setup
- `{mcp_boilerplate}/wrangler.toml` - Cloudflare configuration
💰 **Payment Features:**
- Stripe integration for subscriptions
- Usage-based billing (metered tools)
- User authentication (Google/GitHub OAuth)
- Free vs paid tool tiers
- Cloudflare Workers deployment
🤖 **Agent-MCP (Reference Implementation):**
`{agent_mcp}/` - Additional MCP patterns and examples
📋 **Implementation Patterns:**
1. **Free Tools** - No authentication required
2. **Subscription Tools** - Monthly/annual billing
3. **Metered Usage** - Pay-per-use tools
4. **Premium Features** - Tiered access control
🚀 **Deployment Stack:**
- Cloudflare Workers (serverless)
- Stripe (payment processing)
- OAuth providers (authentication)
- Environment-based configuration
📝 **Getting Started:**
```bash
cd {mcp_boilerplate}
npm install
cp .dev.vars.example .dev.vars
# Configure your Stripe and OAuth keys
npx wrangler dev
```
💡 **Business Model Examples:**
- AI tool marketplace with usage billing
- Premium API access with subscriptions
- Freemium MCP services
- Enterprise authentication integration
🔧 **Customization Guide:**
1. Configure payment providers in `.dev.vars`
2. Implement your tools in `src/tools/`
3. Set up authentication flows
4. Deploy to Cloudflare Workers
5. Monitor usage and billing
"""
except Exception as e:
logger.error(f"Error fetching payment repos: {traceback.format_exc()}")
return f"❌ Error fetching payment repos: {str(e)}\n\nMake sure git is installed and you have internet access."
def main():
"""Main entry point for MCPify MCP server"""
mcp.run()
if __name__ == "__main__":
main()