AI导哈

MCP服务器模板

一个用于构建模型上下文协议（MCP）服务器的最小化、干净的模板，用于与外部服务集成。

  • by Datacolor-Inc
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 datacolor-inc/mcp_server

使用说明

MCP服务器

用于构建与外部服务集成的模型上下文协议（MCP）服务器的最小、干净的模板。此模板演示了核心模式和结构，而没有实际服务连接的复杂性。

🎯 目的

此模板显示：

  • MCP服务器结构：如何组织MCP服务器
  • 工具创建：如何使用适当的文档构建MCP工具
  • 集成模式：外部服务集成的标准方法
  • 错误处理：强大的错误处理和日志记录
  • 配置：如何管理服务配置

📁 包含什么

  • mcp_server.py -使用一个示例工具完成MCP服务器
  • 身份验证流程示例 （模拟）
  • 正确的错误处理和验证
  • 结构化日志记录
  • 清晰的文件和评论

🚀 快速开始

先决条件

pip install fastmcp

运行服务器

python mcp_server.py

🔧 定制

1.更新配置

用您的服务详细信息替换示例配置：

SERVICE_CONFIG = {
    "api_url": "https://your-actual-service.com/api",
    "username": "your_username",
    "api_key": "your_api_key",
    "timeout": 30
}

2.实施身份验证

用实际的身份验证逻辑替换模拟身份验证：

async def authenticate_service():
    try:
        # Replace with actual API call
        response = requests.post(f"{SERVICE_CONFIG['api_url']}/auth", {
            "username": SERVICE_CONFIG["username"],
            "api_key": SERVICE_CONFIG["api_key"]
        })
        
        if response.status_code == 200:
            return {
                "success": True,
                "token": response.json()["access_token"]
            }
        else:
            return {"success": False, "error": "Auth failed"}
            
    except Exception as e:
        return {"success": False, "error": str(e)}

3.更换示例工具

修改 search_resources() 或为您的用例创建新工具：

@mcp.tool()
async def your_custom_tool(param1: str, param2: int = 10) -> Dict[str, Any]:
    """
    Your custom tool description.
    
    Args:
        param1 (str): Description of parameter
        param2 (int): Description with default value
        
    Returns:
        Dict[str, Any]: Description of return value
    """
    try:
        # Your implementation here
        # 1. Authenticate if needed
        # 2. Validate parameters
        # 3. Make API calls
        # 4. Process results
        # 5. Return structured response
        
        return {"success": True, "data": "your_results"}
        
    except Exception as e:
        return {"error": str(e)}

4.添加多个工具

您可以根据需要添加任意数量的工具：

@mcp.tool()
async def create_resource(name: str, data: Dict[str, Any]) -> Dict[str, Any]:
    """Create a new resource."""
    # Implementation here
    pass

@mcp.tool()
async def update_resource(resource_id: str, updates: Dict[str, Any]) -> Dict[str, Any]:
    """Update an existing resource."""
    # Implementation here
    pass

@mcp.tool()
async def delete_resource(resource_id: str) -> Dict[str, Any]:
    """Delete a resource."""
    # Implementation here
    pass

🏗️ 集成示例

REST API集成

import requests

async def call_api(endpoint: str, method: str = "GET", data: Dict = None):
    headers = {"Authorization": f"Bearer {token}"}
    response = requests.request(
        method, 
        f"{SERVICE_CONFIG['api_url']}{endpoint}",
        headers=headers,
        json=data
    )
    return response.json()

数据库集成

import asyncpg

async def query_database(query: str, params: list = None):
    conn = await asyncpg.connect(DATABASE_URL)
    try:
        result = await conn.fetch(query, *params if params else [])
        return [dict(row) for row in result]
    finally:
        await conn.close()

GraphQL集成

import httpx

async def graphql_query(query: str, variables: Dict = None):
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{SERVICE_CONFIG['api_url']}/graphql",
            json={"query": query, "variables": variables},
            headers={"Authorization": f"Bearer {token}"}
        )
        return response.json()

📋 最佳实践

错误处理

try:
    # Your operation
    result = await external_service_call()
    return {"success": True, "data": result}
except ServiceAuthError:
    return {"error": "Authentication failed"}
except ServiceTimeoutError:
    return {"error": "Service timeout"}
except Exception as e:
    logger.error(f"Unexpected error: {e}")
    return {"error": "Internal server error"}

输入验证

if not query or not query.strip():
    return {"error": "Query parameter is required"}

if limit <= 0 or limit > 1000:
    return {"error": "Limit must be between 1 and 1000"}

# Validate format
import re
if not re.match(r'^[a-zA-Z0-9\s]+$', query):
    return {"error": "Invalid query format"}

日志记录

import logging

logger = logging.getLogger(__name__)

# Log important operations
logger.info(f"Processing request: {operation}")
logger.warning(f"Rate limit approaching: {current_usage}")
logger.error(f"Operation failed: {error_details}")

🧪 测试您的实现

基本测试

import asyncio

async def test_tool():
    result = await search_resources("test query", limit=5)
    print(f"Result: {result}")

if __name__ == "__main__":
    asyncio.run(test_tool())

集成测试

async def test_integration():
    # Test authentication
    auth = await authenticate_service()
    assert auth["success"], f"Auth failed: {auth.get('error')}"
    
    # Test tool
    result = await search_resources("integration test")
    assert "error" not in result, f"Tool failed: {result.get('error')}"
    
    print("All tests passed!")

🔒 安全考虑

  • 安全地存储凭据（环境变量、密钥库）
  • 验证所有输入参数
  • 对所有外部通信使用HTTPS
  • 必要时实施限速
  • 适当记录安全事件
  • 小心处理敏感数据

📚 后续步骤

  1. 复制此模板 开始你的项目
  2. 更新配置 提供您的服务详细信息
  3. 实施身份验证 为您的特定服务
  4. 更换示例工具 根据您的业务逻辑
  5. 添加错误处理 针对您的服务
  6. 彻底测试 有真实数据
  7. 添加日志记录 用于监控和调试

此模板为任何MCP服务器集成提供了坚实的基础。专注于您的业务逻辑，同时遵循既定的可靠性和可维护性模式。

相关 MCP