MCP 系列（实战篇）：从可跑通到可上线的 MCP 开发指南

引言

前两篇我们已经讲清楚：

• “为什么要用 MCP”
• “MCP 是怎么通信的”

这一篇只做一件事：把 MCP 真正跑起来，并尽量接近生产实践。

你将看到：

• 三种传输方式下的最小可运行示例（STDIO、SSE、Streamable HTTP）
• 一套可复用的开发最佳实践（Server / Client / 安全 / 测试）
• 常见问题的排查思路（初始化失败、断连、性能）

实战路线图

建议按下面顺序实践：

1. 先跑通 STDIO（本地最简单）
2. 再跑 SSE（理解远程双通道）
3. 最后跑 Streamable HTTP（生产部署首选）
4. 再接入 Agent（让 LLM 自动调用工具）

三种传输方式：最小可运行示例

1）STDIO：本地开发首选

Server 端：

import asyncio
from mcp.server.lowlevel import Server
from mcp.server.stdio import stdio_server

mcp = Server("mysql_mcp_server")

@mcp.list_tools()
async def list_tools():
    return []

@mcp.call_tool()
async def call_tool(name: str, arguments: dict):
    return []

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

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

Client 端：

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

async def main():
    server_params = StdioServerParameters(command="python", args=["mysqlMCPServer.py"])
    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            tools = await session.list_tools()
            print(tools)

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

适用场景：本地调试、单机测试、快速验证业务逻辑。

2）SSE：远程实时推送场景

Server 端：

from fastapi import FastAPI, Request
from mcp.server.sse import SseServerTransport
from starlette.routing import Mount
from mysqlMCPServer import mcp

app = FastAPI()
sse = SseServerTransport("/messages/")
app.router.routes.append(Mount("/messages", app=sse.handle_post_message))

@app.get("/sse")
async def handle_sse(request: Request):
    async with sse.connect_sse(request.scope, request.receive, request._send) as (read_stream, write_stream):
        await mcp.run(read_stream, write_stream, mcp.create_initialization_options())

Client 端：

import asyncio
from mcp import ClientSession
from mcp.client.sse import sse_client

async def main():
    async with sse_client(url="http://localhost:8000/sse") as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            result = await session.call_tool("execute_sql", {"query": "SELECT 1"})
            print(result)

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

适用场景：远程部署、需要服务端主动推送结果或进度。

3）Streamable HTTP：生产环境推荐

Server 端：

from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from starlette.applications import Starlette
from starlette.routing import Mount
from mysqlMCPServer import mcp

session_manager = StreamableHTTPSessionManager(
    app=mcp,
    event_store=None,
    json_response=None,
    stateless=True,
)

async def handle_streamable_http(scope, receive, send):
    await session_manager.handle_request(scope, receive, send)

starlette_app = Starlette(routes=[Mount("/mcp", app=handle_streamable_http)])

Client 端：

import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async def main():
    async with streamablehttp_client(url="http://localhost:8000/mcp") as (read_stream, write_stream, get_session_id):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            print("session_id:", get_session_id())
            result = await session.call_tool("execute_sql", {"query": "SELECT 1"})
            print(result)

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

适用场景：云部署、无状态扩展、网关和中间件友好。

Agent + Streamable HTTP（可执行示例）

下面直接给你一套可复制即跑的代码：一个 Server、一个 Agent Client。

1）Streamable HTTP MCP Server（完整可运行）

import os
import contextlib
import logging
from collections.abc import AsyncIterator

import uvicorn
from dotenv import load_dotenv
from mysql.connector import connect
from mcp.server.lowlevel import Server
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from mcp.types import Tool, TextContent
from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.types import Scope, Receive, Send

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp_streamable_server")

load_dotenv()
HOST = os.getenv("HOST", "0.0.0.0")
PORT = int(os.getenv("PORT", "8000"))

DB_CONFIG = {
    "host": os.getenv("MYSQL_HOST", "127.0.0.1"),
    "user": os.getenv("MYSQL_USER", "root"),
    "password": os.getenv("MYSQL_PASSWORD", ""),
    "database": os.getenv("MYSQL_DATABASE", "test"),
}

mcp = Server("mysql_mcp_server")

@mcp.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="execute_sql",
            description="执行只读 SQL 查询（建议 SELECT）。",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "要执行的 SQL 语句"}
                },
                "required": ["query"],
            },
        )
    ]

@mcp.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name != "execute_sql":
        raise ValueError(f"Unknown tool: {name}")
    query = arguments.get("query", "").strip()
    if not query:
        raise ValueError("query is required")
    if not query.upper().startswith("SELECT"):
        raise ValueError("only SELECT is allowed in this demo")

    with connect(**DB_CONFIG) as conn:
        with conn.cursor() as cursor:
            cursor.execute(query)
            cols = [c[0] for c in cursor.description]
            rows = cursor.fetchall()
            lines = [",".join(cols)] + [",".join(map(str, r)) for r in rows]
            return [TextContent(type="text", text="n".join(lines))]

session_manager = StreamableHTTPSessionManager(
    app=mcp,
    event_store=None,
    json_response=None,
    stateless=True,
)

async def handle_streamable_http(scope: Scope, receive: Receive, send: Send) -> None:
    await session_manager.handle_request(scope, receive, send)

@contextlib.asynccontextmanager
async def lifespan(app: Starlette) -> AsyncIterator[None]:
    async with session_manager.run():
        logger.info("Streamable HTTP MCP Server started")
        yield

app = Starlette(
    routes=[Mount("/mcp", app=handle_streamable_http)],
    lifespan=lifespan,
)

if __name__ == "__main__":
    uvicorn.run(app, host=HOST, port=PORT, log_level="info")

2）Agent Client（完整可运行）

import asyncio
from dotenv import load_dotenv
from langchain.ch@t_models import init_ch@t_model
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent

load_dotenv()

llm = init_ch@t_model(
    model="deepseek-ch@t",
    temperature=0,
    model_provider="deepseek",
)

async def main():
    client = MultiServerMCPClient(
        {
            "mysql_streamable": {
                "url": "http://127.0.0.1:8000/mcp",
                "transport": "streamable_http",
            }
        }
    )

    tools = await client.get_tools()
    agent = create_react_agent(
        model=llm,
        tools=tools,
        prompt=SystemMessage(content="你是数据分析助手，优先调用工具并输出结论。"),
        checkpointer=InMemorySaver(),
    )

    config = {"configurable": {"thread_id": "thread-1"}}
    result = await agent.ainvoke(
        {"messages": [HumanMessage(content="查询 goods 表前 5 条数据，并给出一句摘要")]},
        config=config,
    )
    print(result["messages"][-1].content)

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

3）运行步骤

1. 准备环境变量：DEEPSEEK_API_KEY、MYSQL_HOST、MYSQL_USER、MYSQL_PASSWORD、MYSQL_DATABASE。
2. 先运行上面的 Server 代码。
3. 再运行上面的 Agent Client 代码。

开发最佳实践

1）Server 侧

• 单一职责：一个 MCP Server 专注一个领域。
• 错误可观测：统一异常处理、结构化日志、可追踪 request id。
• 异步优先：I/O 操作全部异步化，数据库连接用连接池。
• 输入校验：对参数做类型、长度、格式校验，尤其是 SQL 与文件路径。

2）Client 侧

• 严格生命周期：统一使用 async with 管理连接和会话。
• 重试与退避：网络错误重试，采用指数退避并设置最大次数。
• 超时治理：连接超时、请求超时都要显式配置。
• 降级策略：远程失败时可降级为本地缓存结果或提示人工介入。

3）安全与权限

• 密钥不要硬编码，统一走环境变量或密钥管理服务。
• 对高风险工具（写库、删文件、发消息）做二次确认。
• 使用最小权限原则，避免 Server 拥有过大系统权限。

4）测试与验收

• 工具级单测：每个 tool 至少覆盖成功、失败、边界三类用例。
• 传输层联调：STDIO、SSE、Streamable HTTP 至少各跑一条回归链路。
• 观测指标：错误率、延迟、超时率、重试率建议纳入坚控。

常见问题与排查

Q1：initialize() 失败怎么办？

排查顺序建议：

1. 先确认 Server 是否启动并正确端点。
2. 再确认传输方式与客户端配置一致（STDIO/SSE/Streamable HTTP）。
3. 查看 Server 日志中的首个异常栈，不要只看客户端报错。
4. 远程模式下检查网络、代理与跨域/网关策略。

Q2：怎么选择传输方式？

• 本地开发：STDIO
• 远程且强调推送：SSE
• 生产部署与扩展性：Streamable HTTP

Q3：连接断开如何处理？

• 增加自动重连与指数退避
• 关键请求做幂等设计
• 重要上下文通过 session 或外部存储持久化

Q4：性能优化从哪里下手？

• 先看慢点是模型推理、网络还是工具执行
• 再做连接池、缓存和并发控制
• 最后才是复杂优化（批处理、分片、异步流水线）

总结

到这里，你已经完成了从“理解 MCP”到“落地 MCP”的闭环：

• 能独立跑通三种传输方式
• 能按工程化方式组织 Server 与 Client
• 能在出问题时快速定位和修复

MCP 的价值，不止是“能调用工具”，更是让 AI 应用具备可维护、可扩展、可上线的工程能力。

参考资源

• MCP 官方简介：www.anthropic.com/news/model-…
• MCP 文档手册：modelcontextprotocol.io/introductio…
• MCP 中文文档：mcp-docs.cn/introductio…
• MCP 官方服务器列表：github.com/modelcontex…
• Python SDK：github.com/modelcontex…