企业级MCP集成方案与应用实践

示例代码

📎 附件：mcp-enterprise.zip（zip, 1316KB）

原文链接：https://www.yuque.com/attachments/yuque/0/2025/zip/2424104/1756179543693-cf7dc8bf-acc4-412c-96c7-9c1fc63b7184.zip

MCP 架构组件介绍

MCP 核心概念

Model Context Protocol (MCP) 是由 Anthropic 于 2024年11月推出的开放标准协议，用于标准化 AI 助手与数据源之间的连接方式。MCP 协议解决了 AI 应用访问外部系统数据时的标准化问题，类似于为 AI 应用提供了一个"USB-C"接口。

核心特性：

• 开放标准：基于 JSON-RPC 的通信协议
• 双向连接：支持客户端和服务器之间的安全双向通信
• 标准化接口：为不同数据源提供统一的访问接口
• 企业级安全：内置安全机制和权限控制

MCP 协议定义了清晰的架构模式：

客户端 (MCP Client)：

• 发起连接和请求的应用程序
• 通常是 AI 应用或 LLM 集成系统
• 负责协议协商和会话管理

服务器 (MCP Server)：

• 提供数据和功能的服务提供方
• 暴露资源、工具和提示词
• 处理客户端请求并返回响应

传输层：

• 支持多种传输方式：stdio、SSE、Streamable HTTP
• 处理消息序列化和网络通信
• 提供连接管理和错误恢复

MCP 协议原语

MCP 定义了五个核心原语：

1. Resources（资源）：只读数据访问接口，类似 GET 端点
2. Tools（工具）：可执行功能接口，类似 POST 端点
3. Prompts（提示词）：可重用的 LLM 交互模板
4. Roots（根）：客户端数据源声明
5. Sampling（采样）：LLM 推理请求处理

版本管理机制

MCP 使用基于日期的版本标识符：

• 格式：YYYY-MM-DD
• 当前版本：2025-06-18
• 版本状态：Draft、Current、Final

最佳实践原则

协议设计最佳实践

1. 版本兼容性评估：选择 AI 提供商时评估 MCP 兼容性
2. 抽象层设计：为多 AI 系统设计抽象层和备用机制
3. 标准传输协议：优先使用官方推荐的传输方式

安全最佳实践

1. 权限最小化原则：仅授予必要的访问权限
2. 连接加密：使用 TLS 加密传输通道
3. 输入验证：严格验证所有输入数据
4. 访问控制：实施细粒度的访问控制机制

# MCP 安全配置示例
mcp_config = {
    "security": {
        "require_tls": True,
        "max_connections": 100,
        "timeout_seconds": 30,
        "allowed_origins": ["https://trusted-domain.com"]
    },
    "access_control": {
        "enable_rbac": True,
        "default_permissions": ["read"],
        "admin_users": ["admin@company.com"]
    }
}

MCP服务器SSE开发实践

环境准备

python版本：3.12.4

# 安装依赖 
mcp>=1.2.0
mcp[cli]
fastmcp>=2.0.0
OpenAI
marvin
psycopg2

基础服务器架构

服务器初始化

from fastmcp import FastMCP
mcp = FastMCP(name="base_mcp_server")

if __name__ == "__main__":
    mcp.run(transport='sse', host="127.0.0.1", port=8001)

资源管理

# 添加一个动态问候资源
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """获取个性化问候"""
    return f"Hello, {name}!"

@mcp.resource("db://province_info/data/{province_name}")
async def get_province_data(province_name: str) -> str:
    """中国省份（含直辖市）信息

    参数:
    province_name: 中国省份名称或直辖市，如广东省，北京市

    返回:
    以 json 格式返回某省份信息
    """
    sql = '''
            select region_code,
                   province_name,
                   capital_city,
                   area,
                   population,
                   abbreviation 
            from public.chinese_provinces 
            where province_name = '{province_name}'
    '''.format(province_name=province_name)
    with get_db_connection() as conn:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute(sql)
            rows = cur.fetchall()
            return json.dumps(list(rows), default=str)