MCP+Agent 工作流

本文档介绍 AI Agent 开发中的核心概念：Function Calling、Agent 智能体、MCP 协议与 A2A 协议。

---

一、Function Calling（函数调用）

1.1 定义与作用

Function Calling 由 OpenAI 等公司推动，允许大语言模型与外部工具连接，将自然语言转换为 API 调用。这解决了大模型在训练结束后知识更新停滞的问题。

通过调用外部工具和服务，Function Calling 帮助大模型解决实时性问题，如"今天温度多少度"、"大盘收盘点数"等。

1.2 工作原理

用户 → LLM识别需求 → 选择函数 → 准备参数 → 调用函数 → 整合回答 → 用户

以天气查询为例：

步骤	描述
第一步	识别需求：这是一个关于实时天气的问题，需要调用外部 API
第二步	选择函数：从可用函数中选择 get_current_weather
第三步	准备参数：{"location": "北京", "unit": "celsius"}
第四步	调用函数：系统使用这些参数调用实际的天气 API
第五步	整合回答："北京今天晴朗，23°C，湿度45%，最高26°C，最低18°C"

1.3 优点与局限

优点	局限
实现简单，适合单一模型少量功能	缺乏跨模型一致性
一键将模型输出对接到代码逻辑	每个 LLM 供应商接口格式略有差异
逻辑直观	无状态性：模型仅生成调用规范，实际执行由外部系统完成

1.4 Function Calling 代码示例

以下是一个使用 OpenAI Function Calling 进行天气查询的完整示例：

import json
from openai import OpenAI

client = OpenAI(api_key="your-api-key")

# 定义可用函数
functions = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "获取指定位置的当前天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "城市名称，如：北京、东京"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位"
                    }
                },
                "required": ["location"]
            }
        }
    }
]

def get_current_weather(location, unit="celsius"):
    """模拟天气查询 API"""
    return f"{location}今天晴朗，23°C，湿度45%"

# 用户请求
messages = [{"role": "user", "content": "北京今天温度多少度？"}]

# 第一次调用：让模型决定是否调用函数
response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=functions
)

assistant_message = response.choices[0].message
print(f"模型回复: {assistant_message}")

# 如果模型选择调用函数
if assistant_message.tool_calls:
    tool_call = assistant_message.tool_calls[0]
    function_name = tool_call.function.name
    function_args = json.loads(tool_call.function.arguments)

    # 执行函数
    if function_name == "get_current_weather":
        result = get_current_weather(**function_args)

    # 将函数结果返回给模型
    messages.append(assistant_message)
    messages.append({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": result
    })

    # 第二次调用：整合结果生成最终回答
    final_response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages
    )
    print(f"最终回答: {final_response.choices[0].message.content}")

---

二、Agent 智能体

2.1 定义

将 Agent 视为人工智能大脑，它使用 LLM 进行推理、计划和采取行动。

• 功能：具备自主决策能力的 AI 系统，通过结合 LLM 的核心推理能力、工具调用（Tools）、记忆（Memory）和规划（Planning）等模块，完成复杂任务。
• 本质：是一个多组件协同的"代理"，能动态选择工具、迭代执行直至目标达成。

2.2 核心特点

特点	说明
自治性	自主决定何时调用工具、如何组合结果
多步迭代	通过"思考-行动-观察"循环处理复杂问题
组件架构	规划模块、工具库、记忆机制协同工作

2.3 组件架构

2.4 执行流程示例

用户请求： "总结 AI 最新进展，用图表展示并邮件给团队。"

步骤	操作
1	调用搜索工具获取最新论文
2	调用摘要生成模型提炼关键点
3	调用图表生成工具创建可视化
4	调用邮件 API 发送结果

2.5 Agent 完整代码示例

以下是基于 LangChain + MCP 工具的 Agent 完整实现：

import asyncio
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

from mcp_demo.agent_client import client_call  # MCP 客户端
from utils.env_utils import OPENAI_API_KEY, OPENAI_BASE_URL

# 初始化模型
model = init_chat_model(
    model="gpt-4o",
    model_provider="openai",
    api_key=OPENAI_API_KEY,
    base_url=OPENAI_BASE_URL
)

# 预加载 MCP 工具（启动时一次性获取）
print('正在获取 MCP 工具...')
_tools = asyncio.run(client_call())
print(f'获取到 {len(_tools)} 个工具')

# 创建 Agent
agent = create_agent(
    model=model,
    tools=_tools,
    system_prompt="你是一个多功能助手"
)

# 调用 Agent 处理用户请求
resp = asyncio.run(agent.ainvoke({
    "messages": [("user", "请计算 10 和 89 的乘积？")]
}))
print('结果:', resp)

---

三、MCP 协议

3.1 定义

MCP（Model Context Protocol，模型上下文协议）由 Anthropic 于 2024 年 11 月推出，是一种开放标准，旨在统一大型语言模型（LLM）与外部数据源和工具之间的通信协议。

三者关系：

• Function Calling 是 AI 模型调用函数的机制
• MCP 是一个标准协议，使 AI 模型与 API 无缝交互
• AI Agent 是一个自主运行的智能系统，利用 Function Calling 和 MCP 来分析和执行任务

3.2 解决的问题

MCP 解决了当前 AI 模型因数据孤岛限制而无法充分发挥潜力的难题：

• 每个新数据源都需要单独定制实现
• 真正互联的系统难以扩展

有了 MCP：

能力	说明
本地数据访问	直接读取本地文件和数据库
互联网数据	访问 Web 和远程 API
开发工具集成	Git、Docker 等开发工具
生产力工具	邮件、日历、文档等
社区生态	各种第三方服务

3.3 工作原理

MCP 协议采用客户端-服务器-资源的三层架构：

3.4 通信机制

MCP 协议支持三种传输机制，均使用 JSON-RPC 2.0 格式进行消息传输：

传输方式	说明	适用场景
stdio	标准输入输出，通过进程 stdin/stdout 进行通信	本地通信，适用于 Claude Desktop 等桌面应用
SSE	Server-Sent Events over HTTP，服务器向客户端推送事件	远程场景，需要服务器主动推送实时更新
streamable_http	HTTP 流式传输，支持请求-响应流和 SSE	远程通信，适用于分布式部署和 Web 应用集成

3.5 MCP 与 Function Calling 的区别

维度	MCP	Function Calling
性质	通信协议标准	函数调用机制
范围	不仅限于 AI 模型，可用于其他应用系统	专为大语言模型设计
灵活性	统一标准，跨模型通用	各模型接口格式略有差异
复杂度	需要完整的客户端-服务器架构	实现简单直接

3.6 MCP 服务器完整代码

使用 FastMCP 框架构建 MCP 服务器，支持工具、资源和提示三大组件：

import mcp
from fastmcp import settings
from fastmcp.tools import tool
from mcp.server import FastMCP
from pydantic import BaseModel, Field
from zai import ZhipuAiClient

from utils.env_utils import ZHIPU_API_KEY

zhipu_client = ZhipuAiClient(api_key=ZHIPU_API_KEY)
mcp_server = FastMCP(
    name='test-mcp',
    instructions='测试的mcp',
)

class SearchInput(BaseModel):
    query: str = Field(description='需要搜索的内容或者关键词')

# ========== 工具 (Tools) ==========
@mcp_server.tool('my_search_tool', description='专门搜索互联网中的内容')
def my_search(query: str) -> str:
    """搜索互联网上的内容"""
    try:
        response = zhipu_client.web_search.web_search(
            search_engine="search-std",
            search_query=query
        )
        if response.search_result:
            return "\n\n".join([d.content for d in response.search_result])
    except Exception as e:
        print(e)
        return '没有搜索到任何内容！'

@mcp_server.tool()
def add(a: int, b: int) -> int:
    """加法运算: 计算两个数字相加"""
    return a + b

@mcp_server.tool()
def multiply(a: int, b: int) -> int:
    """乘法运算：计算两个数字相乘"""
    return a * b

# ========== 资源 (Resources) ==========
@mcp_server.resource("datas://users/{user_id}/email", name='get_user_email')
async def get_user_email(user_id: str) -> str:
    """检索给定用户ID的电子邮件地址。"""
    emails = {"123": "alice@example.com", "456": "bob@example.com"}
    return emails.get(user_id, "not_found@example.com")

@mcp_server.resource("data://product-categories")
async def get_categories() -> list[str]:
    """返回产品分类列表."""
    return ["Electronics", "Books", "Home Goods"]

# ========== 提示 (Prompts) ==========
@mcp_server.prompt("data://product-categories")
async def get_product_categories() -> str:
    """返回产品分类提示."""
    return "Electronics, Books, Home Goods"

# 运行服务器
if __name__ == '__main__':
    mcp_server.run(transport="streamable-http")

3.7 MCP 客户端完整代码

使用 langchain_mcp_adapters 连接远程 MCP 服务器：

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools

# 定义远程 MCP 服务器配置
mcp_server_config = {
    'url': 'http://localhost:8000/mcp',
    'transport': 'streamable-http'
}

async def client_call():
    """客户端：访问服务器中的工具和资源"""
    client = MultiServerMCPClient({
        'test_mcp': mcp_server_config,
    })

    # ========== 工具调用 ==========

    # 方式1：直接通过 client.get_tools() 获取并调用
    tools = await client.get_tools(server_name='test_mcp')
    print('所有可用工具:', tools)

    # 直接调用工具对象
    for tool in tools:
        if tool.name == 'my_search_tool':
            result = await tool.ainvoke({'query': '北京天气'})
            print('方式1-直接调用:', result)

    # 方式2：通过 session 调用单个工具
    async with client.session('test_mcp') as session:
        result = await session.call_tool('my_search_tool', {'query': '北京天气'})
        print('方式2-session调用:', result)

    # ========== 资源获取 ==========

    # 获取指定服务器的所有资源
    all_resources = await client.get_resources(server_name='test_mcp')
    print('所有资源:', all_resources)

    # 获取指定服务器的指定资源
    resources = await client.get_resources(
        server_name='test_mcp',
        uris=['datas://users/123/email']
    )

    # 读取资源内容
    for resource in resources:
        print('资源类型:', resource.metadata)
        print('资源内容:', resource.data if hasattr(resource, 'data') else resource)

    # 通过 session 获取资源
    async with client.session('test_mcp') as session:
        resources = await session.list_resources()
        print('资源列表:', resources)
        resource = await session.read_resource('datas://users/123/email')
        print('单个资源:', resource)

    return tools

if __name__ == '__main__':
    import asyncio
    asyncio.run(client_call())

3.8 MCP 架构流程图

┌─────────────────────────────────────────────────────────────┐
│                        MCP Server                           │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐  │
│  │   Tools     │  │  Resources  │  │      Prompts       │  │
│  │  - search   │  │  - emails   │  │  - product-categories│
│  │  - add      │  │  - categories│ │                     │  │
│  │  - multiply │  │             │  │                     │  │
│  └─────────────┘  └─────────────┘  └─────────────────────┘  │
└────────────────────────┬────────────────────────────────────┘
                         │ streamable-http (localhost:8000)
                         ▼
┌─────────────────────────────────────────────────────────────┐
│                      MCP Client                             │
│  MultiServerMCPClient ──────► 获取工具/资源/调用工具        │
└────────────────────────┬────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────┐
│                    LangChain Agent                          │
│  init_chat_model(gpt-4o) + tools + system_prompt           │
└─────────────────────────────────────────────────────────────┘

---

四、A2A 协议

4.1 概述

A2A（Agent-to-Agent）协议由 Google 于 2025 年 4 月 9 日发布，是一个开放协议，旨在促进 AI Agent 之间的协作，特别适用于大规模、多智能体系统的部署。

4.2 核心原则

原则	说明
拥抱智能体能力	支持自然、非结构化的协作模式
利用现有标准	使用 HTTP、SSE 和 JSON-RPC，确保兼容性
默认安全	支持企业级认证和授权，与 OpenAPI 保持一致
支持长期任务	处理从快速任务到深入研究，提供实时反馈和状态更新
多模态支持	支持文本、音频、视频流等多模态通信

4.3 MCP 与 A2A 的关系

协议	角色定位
MCP	标准化 AI 模型与外部数据源（数据库、API、文件系统）的连接，让 AI 助理能够访问和利用上下文数据
A2A	应用层协议，支持智能体以自然模态协作，允许智能体以"智能体"身份交互，而非作为工具被调用

两者互补，共同推动智能体生态发展。

---

五、FastMCP 快速开发

5.1 简介

FastMCP 是构建 MCP 服务器的快速 Python 方案，提供高层接口、极少样板代码，符合 Python 开发者直觉。

5.2 开发资源

资源	链接
Java 开发 MCP 服务	Spring Boot 改造为 MCP Server
Python 开发 MCP 服务	langchain-mcp-adapters
开源 MCP 服务器	MCP servers | Glama

5.3 主要特性

• 快速：高层接口，更少代码，更快开发
• 简单：所需样板代码最少
• Pythonic：符合 Python 开发者直觉
• 完整：致力于 MCP 规格的全面实现

5.4 FastMCP 代码示例

快速启动服务器

from fastmcp import FastMCP

mcp = FastMCP("我的 MCP 服务器")

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

@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str):
    """获取用户资料"""
    return {"id": user_id, "name": "张三", "email": "zhangsan@example.com"}

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

本地工具模块化

将工具分离到独立文件（如 mcp_tools.py）：

# mcp_tools.py
from mcp_demo.mcp_server import mcp_server

@mcp_server.tool()
def add(a: int, b: int) -> int:
    """加法运算: 计算两个数字相加"""
    return a + b

@mcp_server.tool()
def multiply(a: int, b: int) -> int:
    """乘法运算：计算两个数字相乘"""
    return a * b

然后在服务器中导入：

# mcp_server.py
import mcp_server  # 导入已装饰的工具

if __name__ == '__main__':
    mcp_server.run(transport="streamable-http")

启动脚本

# mcp_tools.py - 直接运行启动
from mcp_demo.mcp_server import mcp_server

if __name__ == '__main__':
    mcp_server.run(transport="streamable-http")

5.5 项目结构示例

一个完整的 MCP Agent 项目结构：

mcp_demo/
├── mcp_server.py      # MCP 服务器主文件
├── mcp_tools.py       # 本地工具定义
├── agent_client.py    # MCP 客户端
├── main.py            # Agent 主程序
└── utils/
    └── env_utils.py   # 环境变量工具

5.6 依赖安装

# 核心依赖
pip install mcp
pip install fastmcp
pip install langchain langchain-openai
pip install langchain-mcp-adapters

# 如使用智谱 AI
pip install zai