MCP协议开发实战：在华为云上搭建AI Agent工具链

1. 引言
TL;DR：面向 Python 开发者，从零掌握 MCP 协议，手把手搭建 MCP Server 与 Agent 客户端，完成「智能运维助手」实战，快速落地可复用的 AI Agent 工具链。

关键词：MCP　AI Agent　Python　Streamable HTTP　工具链

随着大语言模型能力的快速提升，AI Agent 的应用场景越来越广泛。然而，模型本身无法直接访问外部数据源、调用业务系统或操作本地文件，这成为 Agent 落地的核心瓶颈。MCP（Model Context Protocol，模型上下文协议）正是为解决这一问题而诞生的开放标准。

MCP 的价值在于统一了 Agent 与外部工具、数据源的接入方式，避免为每个数据源单独开发定制化集成，让工具能力可复用、可共享、可跨平台迁移。基于这套标准，开发者只需实现一次工具接入，即可被任意支持 MCP 的 AI 应用复用，大幅降低集成成本。

本文将从零开始，手把手带你搭建一套基于 MCP 的 AI Agent 工具链，涵盖协议原理、服务端开发、客户端接入与实战案例，帮助你快速掌握这一关键技术的落地方法。

2. MCP 协议核心概念
2.1 什么是 MCP
MCP 是一种基于 JSON-RPC 2.0 的开放协议，定义了 AI 应用（Host）与外部工具/数据源（Server）之间的标准化通信方式。

2.2 核心角色
Host：AI 应用主体，如 Claude Desktop、自研 Agent 框架

Client：与 Server 建立连接的协议客户端

Server：暴露工具、资源和提示词的服务端### 2.3 三大核心原语

Tools（工具）：可被模型调用的函数，如查询天气、操作数据库

Resources（资源）：可被读取的数据，如文件内容、API 返回

Prompts（提示词）：可复用的提示模板

2.4 通信机制
基于 JSON-RPC 2.0 的消息格式
支持 stdio 与 Streamable HTTP 两种传输方式
会话初始化与能力协商流程
3. 开发环境准备
3.1 技术栈选型
语言：Python 3.10+（本文以 Python 为例）
官方 SDK：mcp Python SDK
框架：FastAPI（用于 HTTP 传输）
3.2 环境搭建
# 创建虚拟环境
python -m venv .venv
source .venv/bin/activate

# 安装 MCP SDK
pip install mcp

# 安装 FastAPI（HTTP 传输需要）
pip install "mcp[fastapi]"
3.3 项目结构规划
mcp-agent-toolchain/
├── server/
│   ├── __init__.py
│   ├── main.py          # 服务端入口
│   ├── tools/           # 工具定义
│   └── resources/       # 资源定义
├── client/
│   ├── __init__.py
│   └── agent.py         # Agent 客户端
└── tests/
4. 开发第一个 MCP Server
4.1 最小服务端实现
# 导入 MCP Server 核心类，用于创建服务端实例
from mcp.server import Server
# 导入 stdio 传输层，负责通过标准输入/输出与客户端通信
from mcp.server.stdio import stdio_server

# 创建 MCP Server 实例，参数 "demo-server" 是服务端名称，用于标识和日志记录
app = Server("demo-server")

# @app.list_tools() 装饰器注册"工具列表"处理器
# 当客户端调用 list_tools 请求时，MCP 框架会自动调用此函数
@app.list_tools()
async def list_tools():
    # 返回工具定义列表，每个工具是一个字典，包含三个关键字段：
    return [
        {
            "name": "get_time",          # 工具名称，客户端通过它来调用
            "description": "获取当前时间", # 工具描述，LLM 据此判断何时调用
            "inputSchema": {              # 输入参数 Schema，定义工具接受的参数结构
                "type": "object",         # 参数必须是 JSON 对象
                "properties": {},         # 该工具无参数，所以属性为空
            },
        }
    ]

# @app.call_tool() 装饰器注册"工具调用"处理器
# 当客户端请求调用某个工具时，MCP 框架会调用此函数
@app.call_tool()
async def call_tool(name: str, arguments: dict):
    # name 参数：客户端请求调用的工具名称
    # arguments 参数：客户端传入的工具参数（字典形式）
    if name == "get_time":
        # 延迟导入 datetime，避免模块加载时的额外开销
        from datetime import datetime
        # 返回 MCP 标准格式的结果，content 是内容列表
        # 每个内容项需指定 type（text 表示文本）和 text（实际内容）
        return {"content": [{"type": "text", "text": str(datetime.now())}]}

# 定义服务端主入口函数
async def main():
    # 使用 async with 打开 stdio 传输通道
    # read 是读取客户端消息的流，write 是向客户端发送消息的流
    async with stdio_server() as (read, write):
        # 启动 MCP Server 主循环，监听并处理来自客户端的请求
        await app.run(read, write)

# 当脚本被直接执行时（而非被导入），运行主函数
if __name__ == "__main__":
    import asyncio
    # 使用 asyncio.run 启动异步事件循环，运行 main() 协程
    asyncio.run(main())
4.2 运行与验证
python server/main.py
4.3 工具注册进阶
使用 @app.tool() 装饰器简化注册
定义带参数的复杂工具
返回结构化数据
5. 构建 Agent 客户端
5.1 客户端连接
from mcp.client.stdio import stdio_client
from mcp import ClientSession

async def connect_to_server():
    server_params = {"command": "python", "args": ["server/main.py"]}
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            return session
5.2 工具发现与调用
通过 session.list_tools() 获取工具列表
通过 session.call_tool() 调用工具
处理工具返回结果
5.3 与 LLM 集成
将工具列表转换为 LLM 可识别的 function calling 格式
模型决策 → 工具调用 → 结果回填的完整循环
6. 实战：搭建完整工具链
6.1 场景设计
构建一个「智能运维助手」，集成以下能力：

查询服务器状态
读取日志文件
执行简单运维命令
6.2 服务端实现
# @app.tool() 装饰器是 MCP 框架提供的便捷注册方式
# 它会自动将函数转换为 MCP 工具，并根据函数签名生成 inputSchema
@app.tool()
async def check_server_status(host: str) -> str:
    """检查服务器状态"""
    # 模拟检查逻辑：实际项目中可替换为真实的 SSH 连接、ping 或 API 调用
    # 返回字符串会被 MCP 自动包装为标准响应格式
    return f"服务器 {host} 运行正常，CPU 使用率 23%"

# 注册第二个工具：读取日志文件
# 函数参数 file_path 和 lines 会自动映射为工具的输入 Schema
@app.tool()
async def read_log(file_path: str, lines: int = 50) -> str:
    """读取日志文件末尾 N 行"""
    # 使用 with 语句安全打开文件，确保文件使用后自动关闭
    with open(file_path, "r") as f:
        # readlines() 读取所有行，[-lines:] 切片取最后 N 行
        content = f.readlines()[-lines:]
    # 将行列表拼接为单个字符串返回
    return "".join(content)
6.3 客户端 Agent 实现
async def run_agent(session: ClientSession, query: str):
    tools = await session.list_tools()
    
    # 将 MCP 工具转换为 LLM function calling 格式
    functions = [
        {
            "name": t.name,
            "description": t.description,
            "parameters": t.inputSchema,
        }
        for t in tools.tools
    ]
    
    # 调用 LLM 进行决策（此处以伪代码示意）
    response = await llm.chat(query, functions=functions)
    
    # 执行工具调用
    if response.tool_calls:
        for call in response.tool_calls:
            result = await session.call_tool(call.name, call.arguments)
            print(f"工具 {call.name} 返回: {result}")
6.4 完整流程演示
是

否

用户输入问题

LLM 决策

需要调用工具?

MCP Client 调用 Server

工具执行并返回结果

结果回填给 LLM

生成最终回答

7. 进阶：Streamable HTTP 传输
7.1 为什么需要 HTTP 传输
支持远程部署与跨网络调用
便于与现有 Web 服务集成
支持多客户端并发访问
7.2 服务端改造
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("remote-server")

@mcp.tool()
def get_weather(city: str) -> str:
    """查询城市天气"""
    return f"{city} 今天晴，25°C"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")
7.3 客户端连接
from mcp.client.streamable_http import streamable_http_client

async with streamable_http_client("http://localhost:8000/mcp") as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()
        tools = await session.list_tools()
8. 安全与最佳实践
8.1 安全注意事项
工具权限最小化原则
输入校验与参数白名单
敏感操作需二次确认
日志脱敏处理
8.2 开发最佳实践
工具命名规范统一
为每个工具编写清晰描述
合理设计输入 Schema
做好错误处理与超时控制
8.3 性能优化
连接复用与长连接
工具结果缓存
异步并发调用
9. 总结与展望
9.1 本文回顾
理解了 MCP 协议的核心概念与通信机制
从零实现了 MCP Server 与 Client
搭建了完整的 AI Agent 工具链
掌握了 HTTP 传输与安全实践
9.2 未来方向
探索 MCP 在更多场景的应用
关注协议版本演进与新特性
构建更复杂的多 Server 协作架构
10. 参考资料
MCP 官方文档：MCP 协议的官方站点，包含协议规范、架构说明与各语言 SDK 的权威指南。
MCP Python SDK 文档：MCP 官方 Python SDK 源码与使用说明，覆盖 Server、Client 及多种传输方式的实现细节。
JSON-RPC 2.0 规范：MCP 底层消息格式所遵循的 JSON-RPC 2.0 官方规范，帮助理解请求、响应与错误对象的结构。
Anthropic MCP 介绍：MCP 协议发布时的官方技术博客，阐述其设计动机与生态愿景。
Model Context Protocol 入门指南：面向初学者的快速上手教程，从概念到第一个 Server 的完整引导。