MCP核心功能

就像快递员帮你跑腿办事一样，MCP在电脑程序之间传递指令和结果。核心三板斧：

• 统一话术：所有工具都用JSON格式对话（类似网购时的订单模板）
• 灵活通道：支持本地直接通话（stdio）和网络远程呼叫（HTTP/2）
• 自动说明书：开发者加个@mcp.tool标签就能生成API文档

MCP客户端并行工具调用流程解析

流程特征说明：

• 并行执行：工具A/B/C同时启动，互不阻塞
• 动态聚合：各工具输出自动合并为结构化数据
• 异步协调：MCP管理任务生命周期和超时控制

服务端实现HELLO world

from mcp.server.fastmcp import FastMCP
mcp = FastMCP("SimpleServer", port=8080, max_connections=1000)  # 扩展连接池容量

@mcp.tool(
    rate_limit=10,
    circuit_breaker=3  # 连续失败3次自动熔断
)
def add_numbers(a: int, b: int) -> int:
    """输入参数验证由Pydantic自动完成"""
    return a + b

if __name__ == "__main__":
    mcp.run(metrics_enabled=True)  # 开启Prometheus指标监控

关键增强点：

• 熔断机制：异常超过阈值时暂停服务
• 连接池管理：预防DDoS攻击
• 监控集成：实时观测QPS/延迟指标

客户端举例

from mcp import ClientSession
from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))
async def call_with_retry(session, payload):
    return await session.call_tool(
        "add_numbers",
        payload,
        compression=True  # 启用传输压缩
    )

async def batch_requests():
    async with ClientSession(
        server_url="ws://cluster.example.com",
        load_balancing=True  # 启用集群负载均衡
    ) as session:
        tasks = [call_with_retry(session, {"a":i, "b":i+1}) for i in range(100)]
        return await asyncio.gather(*tasks)

性能优化技巧：

• 自动重试：处理临时网络故障
• 批量处理：减少RPC调用次数
• 连接复用：保持TCP长连接活性

查询数据库的例子：MCP Server端（Python）

使用FastAPI创建MCP Server，连接MySQL数据库并处理查询请求。

from fastapi import FastAPI, Request  
import pymysql  
from pydantic import BaseModel  

app = FastAPI()  

class Query(BaseModel):  
    query: str  

@app.post("/query")  
async def execute_query(query: Query):  
    connection = pymysql.connect(  
        host='localhost',  
        user='root',  
        password='your_password',  # 建议从环境变量中读取  
        database='your_database',  
        cursorclass=pymysql.cursors.DictCursor  
    )  

    try:  
        with connection.cursor() as cursor:  
            cursor.execute(query.query)  
            result = cursor.fetchall()  
        connection.commit()  
    finally:  
        connection.close()  

    return {"results": result}  

代码实现：MCP Client端（Python）

通过HTTP请求向MCP Server发送SQL查询并获取结果。

import requests  

def query_db(sql_query):  
    url = "http://localhost:8000/query"  
    response = requests.post(url, json={"query": sql_query})  
    if response.status_code == 200:  
        return response.json()['results']  
    else:  
        raise Exception(f"Error querying database: {response.text}")  

# 使用示例  
sql_query = "SELECT * FROM your_table LIMIT 5;"  
try:  
    results = query_db(sql_query)  
    print(results)  
except Exception as e:  
    print(e)  

交互时序图

Mermaid时序图描述MCP Client、Server与MySQL的交互流程。

关键说明

• 安全性：数据库密码通过环境变量管理，避免硬编码。
• 解耦设计：MCP Client无需直接访问数据库，仅与MCP Server交互。
• 扩展性：MCP Server可集成权限控制、查询日志等功能。

总结

MCP框架通过模块化设计实现了高效的工具集成与任务协调，其核心价值体现在标准化通信协议和智能化流程管理。JSON格式的通用数据接口消除了工具间的适配成本，并行调用机制显著提升系统吞吐量。服务端实现的熔断保护与连接池优化保障了高并发下的稳定性，客户端的自动重试与批量处理则优化了交互效率。典型数据库查询案例展示了实际应用场景中的解耦优势：前端无需关注数据库细节，后端可灵活扩展安全审计等能力。时序图清晰呈现了请求流转过程，验证了框架在复杂系统间建立轻量级通信通道的有效性。这种设计模式特别适合需要整合异构系统的AI应用开发场景。