MCP From Scratch

MCP服务器从头开始

从头开始构建模型上下文协议（MCP）服务器的全面指南。此存储库演示了一个完整的MCP服务器实现，其中包含文本摘要和思维导图生成工具，使用stdio传输与MCP客户端进行通信。

什么是MCP？

模型上下文协议（MCP）是一种协议，使AI助手能够安全地访问外部工具、数据源和功能。MCP服务器通过以下方式公开功能：

• 工具：可由AI助手调用的可执行函数
• 资源：可读数据源（文件、数据库、API）
• 提示词：可自定义的基于模板的提示

架构概述

此MCP服务器实现包括：

llm_tools/
├── mcp_server.py          # Core MCP server logic (Tools, Resources, Prompts)
├── mcp_server_main.py     # Stdio transport & JSON-RPC handler
├── llms/                  # LLM provider abstractions
│   ├── base.py           # Base LLM interface
│   ├── factory.py        # LLM factory pattern
│   ├── openai.py         # OpenAI provider
│   ├── ollama.py         # Ollama provider
│   └── vllm.py           # vLLM provider
├── tools/                 # Tool implementations
│   ├── base.py           # Base tool interface
│   ├── factory.py        # Tool factory pattern
│   ├── summary.py        # Text summarization tool
│   └── mindmap.py        # Mindmap generation tool
└── config.json           # Server configuration

构建MCP服务器：一步一步

步骤1：了解MCP协议

MCP通过stdio（stdin/stdout）使用JSON-RPC 2.0。协议流程：

1. 初始化：客户端发送 initialize 请求，服务器以功能响应
2. 已初始化：服务器发送 notifications/initialized 通知
3. 工具列表：客户通过以下方式请求可用工具 tools/list
4. 工具调用：客户通过以下方式调用工具 tools/call
5. 资源/提示：资源和提示的模式类似

步骤2：创建核心MCP服务器类

这 MCPServer 类封装您的工具并通过MCP公开它们：

class MCPServer:
    """MCP Server wrapper for tools."""
    
    def __init__(self, llm_config, tool_configs):
        # Initialize your tools here
        self.summary_tool = ToolFactory.create_tool("summary", llm_config, tool_configs.get("summary", {}))
        self.mindmap_tool = ToolFactory.create_tool("mindmap", llm_config, tool_configs.get("mindmap", {}))
    
    def list_tools(self) -> List[Dict[str, Any]]:
        """Return MCP-compatible tool definitions."""
        return [
            {
                "name": "summarize_text",
                "description": "Generate summary from text",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "text": {"type": "string", "description": "Text to summarize"}
                    },
                    "required": ["text"]
                }
            }
        ]
    
    async def call_tool(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """Execute a tool by name."""
        if name == "summarize_text":
            result = await self.summary_tool.process(arguments)
            return {
                "content": [{"type": "text", "text": json.dumps(result)}],
                "isError": False
            }

步骤3：实施标准传输

MCP服务器通过stdin/stdout进行通信。创建传输处理程序：

class StdioMCPServer:
    """Handles JSON-RPC over stdio."""
    
    async def read_request(self) -> Dict[str, Any]:
        """Read JSON-RPC request from stdin."""
        line_bytes = await asyncio.get_event_loop().run_in_executor(
            None, sys.stdin.buffer.readline
        )
        if not line_bytes:
            return None
        line = line_bytes.decode('utf-8').strip()
        return json.loads(line)
    
    def write_response(self, response: Dict[str, Any]):
        """Write JSON-RPC response to stdout."""
        response_str = json.dumps(response, ensure_ascii=False)
        response_bytes = response_str.encode('utf-8')
        sys.stdout.buffer.write(response_bytes)
        sys.stdout.buffer.write(b'\n')
        sys.stdout.buffer.flush()
    
    async def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
        """Handle JSON-RPC request and return response."""
        method = request.get("method")
        params = request.get("params", {})
        request_id = request.get("id")
        
        if method == "initialize":
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": {
                    "protocolVersion": "2024-11-05",
                    "capabilities": {"tools": {}, "resources": {}, "prompts": {}},
                    "serverInfo": self.mcp_server.get_server_info()
                }
            }
        elif method == "tools/list":
            tools = self.mcp_server.list_tools()
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": {"tools": tools}
            }
        elif method == "tools/call":
            tool_name = params.get("name")
            arguments = params.get("arguments", {})
            result = await self.mcp_server.call_tool(tool_name, arguments)
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": result
            }
        # ... handle other methods

步骤4：实现主循环

服务器在异步循环中运行，读取请求并写入响应：

async def run(self):
    """Run the stdio MCP server."""
    # Send initialized notification
    init_notification = {
        "jsonrpc": "2.0",
        "method": "notifications/initialized"
    }
    self.write_response(init_notification)
    
    # Main loop
    while True:
        try:
            request = await self.read_request()
            if not request:
                break
            
            response = await self.handle_request(request)
            if response:
                self.write_response(response)
        except Exception as e:
            logger.error("Error: %s", e, exc_info=True)
            continue

第五步：定义你的工具

工具是核心功能。每个工具都需要：

1. 架构定义：描述输入的JSON模式
2. 执行逻辑：处理输入的异步函数

工具定义示例：

def list_tools(self) -> List[Dict[str, Any]]:
    tools = []
    
    if self.summary_tool:
        tools.append({
            "name": "summarize_text",
            "description": "Generate abstractive or extractive summary from text",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "text": {
                        "type": "string",
                        "description": "Text content to summarize"
                    },
                    "summary_type": {
                        "type": "string",
                        "enum": ["abstractive", "extractive"],
                        "default": "abstractive"
                    },
                    "language": {
                        "type": "string",
                        "enum": ["vietnamese", "english"],
                        "default": "vietnamese"
                    }
                },
                "required": ["text"]
            }
        })
    
    return tools

步骤6：实施资源（可选）

资源公开可读数据。例子：

def list_resources(self) -> List[Dict[str, Any]]:
    """List available resources."""
    return [
        {
            "uri": "prompt://summary/abstractive/vietnamese",
            "name": "Summary Abstractive Prompt (Vietnamese)",
            "description": "Prompt template for abstractive summarization",
            "mimeType": "text/plain"
        }
    ]

async def read_resource(self, uri: str) -> Dict[str, Any]:
    """Read a resource by URI."""
    if uri.startswith("prompt://"):
        # Parse and return prompt content
        prompt_text = get_prompt(...)
        return {
            "contents": [{
                "uri": uri,
                "mimeType": "text/plain",
                "text": prompt_text
            }]
        }

步骤7：执行提示（可选）

提示是基于模板的提示，可以自定义：

def list_prompts(self) -> List[Dict[str, Any]]:
    """List available prompts."""
    return [
        {
            "name": "summarize_abstractive",
            "description": "Generate abstractive summary from text",
            "arguments": [
                {
                    "name": "text",
                    "description": "Text content to summarize",
                    "required": True
                }
            ]
        }
    ]

async def get_prompt(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
    """Get a prompt template with arguments filled in."""
    if name == "summarize_abstractive":
        text = arguments.get("text", "")
        language = arguments.get("language", "vietnamese")
        
        system_prompt = get_prompt("summary", language, "abstractive")
        user_prompt = f"Summarize: {text}"
        
        return {
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ]
        }

安装

先决条件

• Python 3.8+
• 点

再进行

pip install -r requirements.txt

可选依赖

为了增强功能：

# Token counting
pip install tiktoken>=0.5.0

# Transformers (for Qwen models)
pip install transformers>=4.30.0 torch>=2.0.0

# LangChain (for advanced text splitting)
pip install langchain>=0.1.0 langchain-text-splitters>=0.0.1

# Ollama support
pip install ollama>=0.1.0 requests>=2.31.0

# vLLM support
pip install vllm>=0.2.0

配置

创建配置文件

创建 config.json:

{
  "llm_config": {
    "enabled": true,
    "provider": "openai",
    "model_name": "gpt-4",
    "api_key": "your_api_key",
    "base_url": "https://api.openai.com/v1",
    "default_temperature": 0.7,
    "default_max_tokens": 2000
  },
  "tool_configs": {
    "summary": {
      "summary_type": "abstractive",
      "language": "english",
      "max_length": 2000
    },
    "mindmap": {
      "language": "english",
      "max_nodes": 50,
      "max_depth": 4
    }
  }
}

环境变量（备选）

创建 .env 文件：

OPENAI_API_KEY=your_api_key_here
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=gpt-4

运行MCP服务器

基本用法

# With default config (Ollama)
python mcp_server_main.py

# With config file
python mcp_server_main.py --config config.json

# With settings module
python mcp_server_main.py --settings-module app.settings

测试服务器

python test_mcp_server.py

这将：

• 初始化服务器
• 列出可用工具
• 测试 summarize_text 工具
• 测试 create_mindmap 工具
• 列出资源

与Claude Desktop集成

macOS

编辑 ~/Library/Application Support/Claude/claude_desktop_config.json:

{
  "mcpServers": {
    "intramind-tools": {
      "command": "python",
      "args": [
        "/absolute/path/to/llm_tools/mcp_server_main.py",
        "--config",
        "/absolute/path/to/config.json"
      ]
    }
  }
}

视窗

编辑 %APPDATA%\Claude\claude_desktop_config.json:

{
  "mcpServers": {
    "intramind-tools": {
      "command": "python",
      "args": [
        "E:\\path\\to\\llm_tools\\mcp_server_main.py",
        "--config",
        "E:\\path\\to\\config.json"
      ],
      "env": {
        "OPENAI_API_KEY": "your_api_key"
      }
    }
  }
}

配置后重新启动Claude Desktop。

使用MCP SDK（替代方案）

您还可以使用官方的MCP SDK：

pip install mcp

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def use_mcp_server():
    server_params = StdioServerParameters(
        command="python",
        args=["mcp_server_main.py", "--config", "config.json"]
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # List tools
            tools = await session.list_tools()
            print("Tools:", tools)
            
            # Call tool
            result = await session.call_tool(
                "summarize_text",
                {"text": "Your text here..."}
            )
            print("Result:", result)

asyncio.run(use_mcp_server())

关键概念

JSON-RPC 2.0协议

MCP使用JSON-RPC 2.0。请求格式：

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "summarize_text",
    "arguments": {"text": "..."}
  }
}

响应格式：

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [{"type": "text", "text": "..."}],
    "isError": false
  }
}

标准运输

• 无HTTP：MCP使用stdin/stdout，而不是HTTP
• 无端口：无需打开端口或配置防火墙
• 基于流程：每个客户端创建自己的服务器进程
• UTF-8编码：始终对文本使用UTF-8编码

错误处理

MCP错误响应格式：

{
  "jsonrpc": "2.0",
  "id": 1,
  "error": {
    "code": -32603,
    "message": "Internal error: ..."
  }
}

标准错误代码：

• -32700：分析错误
• -32600：无效请求
• -32601：未找到方法
• -32602：无效参数
• -32603：内部错误

项目结构说明

mcp_server.py

核心MCP服务器实现：

• MCPServer 类：包装工具并公开MCP接口
• list_tools()：返回工具定义
• call_tool()：执行工具
• list_resources() / read_resource()：资源管理
• list_prompts() / get_prompt()：提示模板

mcp_server_main.py

标准运输和入境点：

• StdioMCPServer 类：通过stdio处理JSON-RPC
• read_request() / write_response()：I/O操作
• handle_request()：将请求路由到处理程序
• main()：加载配置的入口点

工具系统

工具在 tools/:

• BaseTool：抽象基类
• SummaryTool：文本摘要
• MindmapTool：思维导图生成
• 工具创建的工厂模式

LLM系统

LLM提供商 llms/:

• BaseLLM：抽象接口
• OpenAILLM：OpenAI API
• OllamaLLM：当地Olama
• vLLMLLM：本地vLLM
• 供应商选择的工厂模式

调试

启用调试日志记录

编辑 mcp_server_main.py:

logging.basicConfig(
    level=logging.DEBUG,  # Change from INFO to DEBUG
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

检查服务器日志

服务器日志转到stderr。在克劳德桌面中：

• macOS: ~/Library/Logs/Claude/
• 视窗: %APPDATA%\Claude\logs\

常见问题

“找不到命令”

• 确保Python在PATH中
• 或者使用完整路径： "C:\\Python39\\python.exe"

“找不到模块”

• 安装依赖项： pip install -r requirements.txt
• 检查Python环境

“连接被拒绝”

• MCP不是HTTP服务器
• 无需端口-使用stdio
• 检查配置文件格式

服务器未响应

• 检查配置中的JSON格式
• 验证API密钥
• 启用调试日志记录

扩展服务器

添加新工具

1. 在中创建工具类 tools/:

class MyTool(BaseTool):
    async def process(self, input_data, **kwargs):
        # Your logic here
        return {"result": "..."}

1. 注册 MCPServer.list_tools():

tools.append({
    "name": "my_tool",
    "description": "My tool description",
    "inputSchema": {...}
})

1. 在中添加处理程序 MCPServer.call_tool():

if name == "my_tool":
    return await self._call_my_tool(arguments)

添加新的LLM提供程序

1. 在中创建提供程序类 llms/:

class MyLLM(BaseLLM):
    async def generate(self, prompt, **kwargs):
        # Your logic here
        return response

1. 注册 LLMFactory:

if provider == "my_llm":
    return MyLLM(config)

最佳实践

1. 始终验证输入：处理前检查工具参数
2. 优雅地处理错误：返回正确的错误响应
3. 使用async/await：MCP操作应该是异步的
4. 适当记录：使用日志进行调试，而不是stdout
5. UTF-8编码：文本I/O始终使用UTF-8
6. 临时工具：工具应该可以安全重试
7. 架构验证：使用JSON模式进行输入验证
8. 资源清理：在错误情况下清理资源

测试

手动测试

使用测试脚本：

python test_mcp_server.py

单元测试

测试单个组件：

import pytest
from mcp_server import MCPServer

async def test_summarize_tool():
    server = MCPServer(llm_config, tool_configs)
    result = await server.call_tool(
        "summarize_text",
        {"text": "Test text"}
    )
    assert "content" in result

资源

• MCP规范
• JSON-RPC 2.0规范
• 克劳德桌面MCP指南

许可证

此项目根据MIT许可证获得许可-请参阅 许可证.md 文件以获取详细信息。

版权所有（c）2025 Duy Ta Khanh

贡献

欢迎投稿！请提交pull请求或bug和功能请求的开放问题。