Python MCP Server开发：2026年从零构建AI Agent可调用的工具服务

AI与大数据

Python MCP Server开发：2026年从零构建AI Agent可调用的工具服务

你在开发AI Agent时是不是发现：每个AI平台都有自己的工具调用协议，Claude用一套，ChatGPT用另一套，Gemini又不一样？你的工具代码要为每个平台写一遍适配层？2026年，Model Context Protocol（MCP）已经成为AI工具调用的开放标准——写一次MCP Server，所有主流AI平台都能调用。

背景知识

MCP协议架构

MCP（Model Context Protocol）是Anthropic在2024年底发布的开放协议，定义了AI应用（Host）与工具服务（Server）之间的标准通信方式。

概念 说明 类比
Host AI应用（Claude Desktop、ChatGPT等） 浏览器
Client Host内部的MCP客户端 HTTP Client
Server 提供工具/资源/提示的服务 HTTP Server
Transport 通信方式（stdio/SSE/Streamable HTTP） TCP/HTTP

MCP核心能力

能力 说明 示例
Tools 可被AI调用的函数 查询数据库、发送邮件
Resources 可被AI读取的数据 文件内容、数据库记录
Prompts 预定义的提示模板 代码审查模板
Sampling AI向Host请求生成 需要AI判断的中间步骤

传输方式对比

传输 适用场景 优点 缺点
stdio 本地桌面应用 简单、安全 仅限本地进程
SSE 远程Web服务 跨网络 长连接维护
Streamable HTTP 远程Web服务（推荐） 无状态、CDN友好 需MCP 2025-03规范

问题分析

为什么需要MCP？

  1. 协议碎片化：每个AI平台的工具调用格式不同，适配成本高
  2. 工具复用难：为Claude写的工具无法在ChatGPT中使用
  3. 安全风险：直接暴露API端点给AI，缺乏权限控制
  4. 状态管理混乱：AI与工具之间的上下文传递没有标准

MCP通过统一协议解决这些问题：一个MCP Server可以被任何支持MCP的AI Host调用。

分步实操

步骤1：安装MCP Python SDK

pip install mcp[cli]

步骤2：创建第一个MCP Server

# server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

app = Server("toolsku-tools")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="calculate",
            description="执行数学表达式计算",
            inputSchema={
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如 '2 + 3 * 4'",
                    }
                },
                "required": ["expression"],
            },
        ),
        Tool(
            name="format_json",
            description="格式化JSON字符串",
            inputSchema={
                "type": "object",
                "properties": {
                    "json_str": {
                        "type": "string",
                        "description": "待格式化的JSON字符串",
                    },
                    "indent": {
                        "type": "integer",
                        "description": "缩进空格数，默认2",
                        "default": 2,
                    },
                },
                "required": ["json_str"],
            },
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "calculate":
        try:
            result = eval(arguments["expression"], {"__builtins__": {}}, {})
            return [TextContent(type="text", text=f"计算结果: {result}")]
        except Exception as e:
            return [TextContent(type="text", text=f"计算错误: {str(e)}")]

    elif name == "format_json":
        import json
        try:
            parsed = json.loads(arguments["json_str"])
            indent = arguments.get("indent", 2)
            formatted = json.dumps(parsed, indent=indent, ensure_ascii=False)
            return [TextContent(type="text", text=formatted)]
        except json.JSONDecodeError as e:
            return [TextContent(type="text", text=f"JSON解析错误: {str(e)}")]

    raise ValueError(f"Unknown tool: {name}")

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

步骤3：配置Claude Desktop连接

// ~/Library/Application Support/Claude/claude_desktop_config.json (macOS)
// %APPDATA%\Claude\claude_desktop_config.json (Windows)
{
  "mcpServers": {
    "toolsku-tools": {
      "command": "python",
      "args": ["server.py"],
      "cwd": "E:\\project\\toolsku\\mcp-server"
    }
  }
}

步骤4：SSE传输模式（远程部署）

# sse_server.py
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp.types import Tool, TextContent
from starlette.applications import Starlette
from starlette.routing import Mount, Route

app = Server("toolsku-remote-tools")
sse = SseServerTransport("/messages/")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_docs",
            description="搜索技术文档",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"},
                    "category": {"type": "string", "description": "文档分类"},
                    "limit": {"type": "integer", "description": "返回数量", "default": 5},
                },
                "required": ["query"],
            },
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "search_docs":
        results = await search_documents(
            query=arguments["query"],
            category=arguments.get("category"),
            limit=arguments.get("limit", 5),
        )
        import json
        return [TextContent(type="text", text=json.dumps(results, ensure_ascii=False, indent=2))]
    raise ValueError(f"Unknown tool: {name}")

async def search_documents(query: str, category: str | None, limit: int) -> list[dict]:
    return [
        {"title": f"文档: {query}", "url": f"https://docs.example.com/{query}", "score": 0.95},
    ][:limit]

async def handle_sse(request):
    async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

starlette_app = Starlette(
    routes=[
        Route("/sse", endpoint=handle_sse),
        Mount("/messages/", app=sse.handle_post_message),
    ],
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(starlette_app, host="0.0.0.0", port=8080)

步骤5：与ChatGPT集成

# chatgpt_adapter.py
import httpx
from openai import OpenAI

MCP_SERVER_URL = "http://localhost:8080"

async def discover_mcp_tools() -> list[dict]:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{MCP_SERVER_URL}/messages/",
            json={
                "jsonrpc": "2.0",
                "method": "tools/list",
                "id": 1,
            },
        )
        result = response.json()
        return result.get("result", {}).get("tools", [])

def mcp_tools_to_openai_tools(mcp_tools: list[dict]) -> list[dict]:
    openai_tools = []
    for tool in mcp_tools:
        openai_tools.append({
            "type": "function",
            "function": {
                "name": tool["name"],
                "description": tool["description"],
                "parameters": tool["inputSchema"],
            },
        })
    return openai_tools

async def call_mcp_tool(name: str, arguments: dict) -> str:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{MCP_SERVER_URL}/messages/",
            json={
                "jsonrpc": "2.0",
                "method": "tools/call",
                "params": {"name": name, "arguments": arguments},
                "id": 2,
            },
        )
        result = response.json()
        contents = result.get("result", {}).get("content", [])
        return "\n".join(c.get("text", "") for c in contents)

async def chat_with_tools(user_message: str) -> str:
    mcp_tools = await discover_mcp_tools()
    openai_tools = mcp_tools_to_openai_tools(mcp_tools)

    client = OpenAI()
    messages = [{"role": "user", "content": user_message}]

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=openai_tools,
    )

    message = response.choices[0].message

    if message.tool_calls:
        for tool_call in message.tool_calls:
            tool_result = await call_mcp_tool(
                tool_call.function.name,
                json.loads(tool_call.function.arguments),
            )
            messages.append(message)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": tool_result,
            })

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=openai_tools,
        )
        return response.choices[0].message.content

    return message.content

完整代码：生产级MCP Server

# production_server.py
import os
import json
import logging
from datetime import datetime
from typing import Any

from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp.types import Tool, TextContent, ImageContent
from starlette.applications import Starlette
from starlette.routing import Mount, Route
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("toolsku-mcp")

app = Server("toolsku-production")

API_KEY = os.environ.get("MCP_API_KEY", "dev-key")

def validate_api_key(headers: dict) -> bool:
    auth = headers.get("authorization", "")
    return auth == f"Bearer {API_KEY}"

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="base64_encode",
            description="Base64编码字符串",
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "待编码的文本"},
                },
                "required": ["text"],
            },
        ),
        Tool(
            name="base64_decode",
            description="Base64解码字符串",
            inputSchema={
                "type": "object",
                "properties": {
                    "encoded": {"type": "string", "description": "Base64编码的字符串"},
                },
                "required": ["encoded"],
            },
        ),
        Tool(
            name="json_format",
            description="格式化JSON字符串",
            inputSchema={
                "type": "object",
                "properties": {
                    "json_str": {"type": "string", "description": "JSON字符串"},
                    "indent": {"type": "integer", "description": "缩进", "default": 2},
                    "sort_keys": {"type": "boolean", "description": "是否排序键", "default": False},
                },
                "required": ["json_str"],
            },
        ),
        Tool(
            name="hash_compute",
            description="计算文本的哈希值",
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "待计算的文本"},
                    "algorithm": {
                        "type": "string",
                        "enum": ["md5", "sha1", "sha256", "sha512"],
                        "description": "哈希算法",
                        "default": "sha256",
                    },
                },
                "required": ["text"],
            },
        ),
        Tool(
            name="timestamp_convert",
            description="时间戳与日期格式互转",
            inputSchema={
                "type": "object",
                "properties": {
                    "value": {"type": "string", "description": "时间戳或日期字符串"},
                    "format": {"type": "string", "description": "目标格式", "default": "iso"},
                },
                "required": ["value"],
            },
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    logger.info(f"Tool called: {name} with args: {arguments}")

    try:
        if name == "base64_encode":
            import base64
            result = base64.b64encode(arguments["text"].encode()).decode()
            return [TextContent(type="text", text=result)]

        elif name == "base64_decode":
            import base64
            result = base64.b64decode(arguments["encoded"]).decode()
            return [TextContent(type="text", text=result)]

        elif name == "json_format":
            parsed = json.loads(arguments["json_str"])
            indent = arguments.get("indent", 2)
            sort_keys = arguments.get("sort_keys", False)
            formatted = json.dumps(parsed, indent=indent, sort_keys=sort_keys, ensure_ascii=False)
            return [TextContent(type="text", text=formatted)]

        elif name == "hash_compute":
            import hashlib
            algorithm = arguments.get("algorithm", "sha256")
            h = hashlib.new(algorithm)
            h.update(arguments["text"].encode())
            return [TextContent(type="text", text=h.hexdigest())]

        elif name == "timestamp_convert":
            value = arguments["value"]
            fmt = arguments.get("format", "iso")
            try:
                ts = float(value)
                dt = datetime.fromtimestamp(ts)
            except ValueError:
                dt = datetime.fromisoformat(value)
                ts = dt.timestamp()

            if fmt == "iso":
                result = dt.isoformat()
            elif fmt == "timestamp":
                result = str(ts)
            else:
                result = dt.strftime(fmt)

            return [TextContent(type="text", text=result)]

        raise ValueError(f"Unknown tool: {name}")

    except Exception as e:
        logger.error(f"Tool error: {name} - {str(e)}")
        return [TextContent(type="text", text=f"错误: {str(e)}")]

sse = SseServerTransport("/messages/")

async def handle_sse(request):
    async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

middleware = [
    Middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]),
]

starlette_app = Starlette(
    routes=[
        Route("/sse", endpoint=handle_sse),
        Mount("/messages/", app=sse.handle_post_message),
    ],
    middleware=middleware,
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(starlette_app, host="0.0.0.0", port=8080)

避坑指南

坑1：Claude Desktop无法发现MCP Server

现象：配置了claude_desktop_config.json但Claude Desktop看不到工具。

解决：检查JSON格式是否正确（注意逗号和引号），确认command路径正确（使用绝对路径），查看Claude Desktop日志（Help > Toggle Developer Tools）。

坑2：SSE连接频繁断开

现象：远程MCP Server的SSE连接每隔30秒断开。

解决：确保服务端实现了心跳机制（每15秒发送keep-alive），配置反向代理的超时时间大于60秒。Nginx需要proxy_read_timeout 300s。

坑3：工具参数校验不严格导致运行时错误

现象：AI传入了错误类型的参数（如字符串传给了需要整数的参数）。

解决：在inputSchema中严格定义类型和约束（type、minimum、enum等），在call_tool函数中添加参数校验和类型转换。

坑4：stdio模式下Python环境不一致

现象：Claude Desktop启动MCP Server时找不到已安装的包。

解决：在claude_desktop_config.json中使用Python解释器的绝对路径，或使用uv run指定虚拟环境：

{
  "mcpServers": {
    "toolsku-tools": {
      "command": "uv",
      "args": ["run", "--directory", "E:\\project\\toolsku\\mcp-server", "server.py"]
    }
  }
}

坑5：并发请求导致状态混乱

现象：多个AI同时调用同一个MCP Server，共享状态被覆盖。

解决：MCP Server应设计为无状态，所有状态通过参数传递。如果必须有状态，使用请求级别的上下文隔离，避免全局变量。

报错排查

序号 报错信息 原因 解决方法
1 MCP server not found in config 配置文件路径或格式错误 检查JSON格式，使用绝对路径
2 Connection refused on stdio Python进程启动失败 检查command路径，查看stderr日志
3 SSE connection timeout 服务端未响应或网络问题 检查服务端状态，增加超时配置
4 Tool not found: xxx 工具名拼写错误 检查list_tools和call_tool的名称一致
5 JSON-RPC error: -32600 请求格式不符合JSON-RPC规范 检查请求body格式
6 Permission denied 工具执行需要权限 检查文件/网络权限，使用sudo谨慎处理
7 ModuleNotFoundError Python依赖未安装 在MCP配置中指定正确的Python环境
8 413 Payload Too Large 请求body过大 限制工具参数大小，使用分页
9 CORS error SSE跨域请求被拦截 配置CORS中间件
10 Server initialization failed MCP握手失败 检查initialize响应格式和capabilities

进阶优化

1. 资源（Resources）暴露

@app.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="toolsku:///docs/api-reference",
            name="API参考文档",
            mimeType="text/markdown",
        ),
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "toolsku:///docs/api-reference":
        return "# API参考\n\n## 工具列表\n..."
    raise ValueError(f"Unknown resource: {uri}")

2. 提示模板（Prompts）

@app.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="code-review",
            description="代码审查提示模板",
            arguments=[
                PromptArgument(name="language", description="编程语言", required=True),
                PromptArgument(name="code", description="待审查的代码", required=True),
            ],
        ),
    ]

@app.get_prompt()
async def get_prompt(name: str, arguments: dict) -> list[PromptMessage]:
    if name == "code-review":
        return [
            PromptMessage(
                role="user",
                content=TextContent(
                    type="text",
                    text=f"请审查以下{arguments['language']}代码，关注安全性、性能和可维护性：\n\n```{arguments['language']}\n{arguments['code']}\n```",
                ),
            ),
        ]
    raise ValueError(f"Unknown prompt: {name}")

3. Streamable HTTP传输（MCP 2025-03）

from mcp.server.streamable_http import StreamableHTTPTransport

http_transport = StreamableHTTPTransport("/mcp")

async def handle_http(request):
    await http_transport.handle_request(request, app)

4. 认证与限流

from starlette.middleware.base import BaseHTTPMiddleware

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        if request.url.path in ["/sse", "/messages/"]:
            auth = request.headers.get("authorization", "")
            if auth != f"Bearer {API_KEY}":
                from starlette.responses import JSONResponse
                return JSONResponse({"error": "unauthorized"}, status_code=401)
        return await call_next(request)

对比分析

维度 MCP OpenAI Function Calling LangChain Tools LlamaIndex Tools
协议标准 开放标准 平台私有 框架私有 框架私有
跨平台 所有支持MCP的Host 仅OpenAI 仅LangChain 仅LlamaIndex
传输方式 stdio/SSE/HTTP HTTP API Python函数 Python函数
工具发现 动态list_tools 静态定义 静态定义 静态定义
资源暴露 原生支持 不支持 不支持 不支持
安全模型 权限控制 API Key 无标准 无标准
部署方式 本地/远程 远程 进程内 进程内
生态 快速增长 成熟 成熟 成熟

总结展望

总结：MCP在2026年已成为AI工具调用的开放标准。Python MCP SDK让开发一个MCP Server变得极其简单——定义工具、注册处理器、选择传输方式。核心优势是"一次开发，多平台调用"：同一个MCP Server可以被Claude Desktop、ChatGPT、Cursor等任何支持MCP的AI应用使用。建议从简单的工具服务入手（如JSON格式化、Base64编解码），逐步添加资源和提示模板能力，最终构建完整的AI工具生态。

在线工具推荐

本站提供浏览器本地工具，免注册即可试用 →

#Python#MCP#Model Context Protocol#AI工具#SSE#stdio#Agent#工具服务