摘要： 本文深入SmartVoyage智途v7.8的A2A协议层与MCP工具层，从AgentCard能力声明、A2AServer任务状态机，到MCP工具注册、LangChain Agent桥接、以及底层SQL/向量检索的完整链路。全文基于源码逐行展开，揭示三个子Agent（天气/票务/行程）是如何实现"配置驱动、即插即用"的。

关键词： A2A协议、MCP协议、AgentCard、FastMCP、LangChain Agent、Streamable HTTP

---

一、引言

前两篇文章分别拆解了系统的六层架构全景和ReAct推理核心。但有一个关键问题还没有展开——当主Agent通过A2A协议把一个任务下发给子Agent后，子Agent内部到底发生了什么？

从收到一句"帮我查明天北京到上海的机票"，到最终返回结构化的航班数据，中间经历了A2A协议解析、MCP工具发现、LangChain Agent推理、SQL查询执行等多个环节。

本文将以天气、票务、行程三个子Agent的源码为基础，完整还原这条链路。

---

二、A2A协议：Agent世界的"外交准则"

2.1 什么是A2A协议？

A2A（Agent-to-Agent）协议是Google在2025年提出的智能体间标准化通信协议。它的核心思想是：每个Agent都是一个独立的服务，通过标准化的消息格式互相协作，而不是把所有逻辑塞进一个巨大的单体Agent中。

在SmartVoyage中，A2A协议承担的角色类似于微服务架构中的RPC框架——主Agent不需要知道子Agent内部怎么实现，只需要知道"它能干什么"和"怎么联系它"。

2.2 两个核心概念

AgentCard（代理名片）——每个Agent的"身份证"，声明了：

• 我是谁（name、description）
• 我能干什么（skills列表）
• 怎么联系我（url）
• 我支持什么能力（capabilities：是否支持流式、是否有记忆）

TaskState（任务状态机）——每个任务的生命周期：

Submitted → InProgress → Completed
                       → InputRequired（需要用户补充信息）
                       → Failed（执行失败）

---

三、AgentCard：让主Agent"认识"你

3.1 天气Agent的名片

# a2a_server/a2a_weather.py
weather_agent_card = AgentCard(
    name=CONFIG.A2A_WEATHER_NAME,       # "WeatherQueryAssistant"
    description="基于LangChain+MCP(Streamable HTTP)提供自然语言天气查询的独立子代理",
    url=A2A_SELF_URL,                    # "http://a2a-weather:6001"
    capabilities={"streaming": False, "memory": False},
    skills=[
        AgentSkill(
            name="query weather",
            description="解析自然语言查询天气（历史预报+实时天气），支持单日/多日范围，信息缺失时主动追问用户",
            examples=["郑州今天天气", "北京未来5天天气", "上海现在实时天气"]
        )
    ],
)

3.2 票务Agent的名片（6个技能）

# a2a_server/a2a_ticket.py
ticket_agent_card = AgentCard(
    name=CONFIG.A2A_TICKET_NAME,         # "TicketAssistant"
    description="基于LangChain+MCP提供火车票/机票/演唱会票查询与预定的独立子代理",
    url=A2A_SELF_URL,
    capabilities={"streaming": False, "memory": False},
    skills=[
        AgentSkill(name="query train tickets",
                   description="查询火车票信息，支持按出发/到达城市、日期、座位类型筛选",
                   examples=["北京到上海的高铁 2026-05-01"]),
        AgentSkill(name="query flight tickets",
                   description="查询航班机票信息，支持按出发/到达城市、日期、舱位筛选",
                   examples=["上海到北京的机票 5月1日"]),
        AgentSkill(name="query concert tickets",
                   description="查询演唱会门票信息，支持按城市、艺人、日期、票型筛选",
                   examples=["薛之谦成都演唱会"]),
        AgentSkill(name="order train tickets",
                   description="预定火车票",
                   examples=["帮我订一张G351二等座 5月1日"]),
        AgentSkill(name="order flight tickets",
                   description="预定机票",
                   examples=["订一张CA4116商务舱 5月1日"]),
        AgentSkill(name="order concert tickets",
                   description="预定演唱会票",
                   examples=["买2张薛之谦VIP内场 5月1日"]),
    ]
)

3.3 名片如何被主Agent使用？

在第二篇文章中我们提到，主Agent的VoyageChatCore在初始化时会遍历所有注册的子Agent，收集它们的AgentCard：

# core/chat_service_core.py — _load_intent_info()
for agent_name in self._network.agents.keys():
    card = self._network.get_agent_card(agent_name)
    for skill in card.skills:
        intent_agentname_dict[skill.name] = agent_name   # 路由映射
        skillname_info_dict[skill.name] = skill.description  # 意图描述

最终，AgentCard中的skills信息被注入到意图识别的Prompt中，让LLM知道"有哪些工具可用"。

这就是"配置驱动"的核心：新增一个Agent只需在config.yaml中添加配置并部署，主Agent会自动发现它的能力并纳入路由。

---

四、A2AServer：任务状态机的实现

4.1 handle_task：核心回调方法

每个子Agent都继承A2AServer，重写handle_task()方法。以天气Agent为例：

# a2a_server/a2a_weather.py
class WeatherQueryA2AServer(A2AServer):
    def __init__(self):
        super().__init__(agent_card=weather_agent_card)

    def handle_task(self, task):
        query = task.message['content']['text']
        query_result = asyncio.run(query_weather(query))
        reply = query_result["message"]

        if query_result["status"] == "success":
            if "请提供" in reply:
                # 需要用户补充信息 → INPUT_REQUIRED
                task.status = TaskStatus(
                    state=TaskState.INPUT_REQUIRED,
                    message={"role": "agent", "content": {"text": reply}}
                )
            else:
                # 查询完成 → COMPLETED，结果放入artifacts
                task.artifacts = [{"parts": [{"type": "text", "text": reply}]}]
                task.status = TaskStatus(state=TaskState.COMPLETED)
        elif query_result["status"] == "error":
            # 查询失败 → FAILED
            task.status = TaskStatus(
                state=TaskState.FAILED,
                message={"role": "agent", "content": {"text": reply}}
            )
        return task

4.2 三种任务状态

图1：A2A任务状态机——三种终态对应三种业务场景

三种状态的业务含义：

状态	触发条件	主Agent的处理方式
Completed	查询成功，结果放入artifacts	提取artifacts[0]['parts'][0]['text']作为结果
InputRequired	回复中包含"请提供"等追问语	将追问消息直接返回给用户
Failed	异常或无数据	返回"暂不支持此意图"兜底回复

4.3 主Agent如何解析TaskResponse？

回到第二篇文章中的_run_single_intent()方法：

# core/chat_service_core.py
raw_response = await asyncio.wait_for(
    agent.send_task_async(task), timeout=150)

if raw_response.status.state == 'completed' and raw_response.artifacts:
    agent_result = raw_response.artifacts[0]['parts'][0]['text']
elif raw_response.status.state == 'input-required':
    agent_result = raw_response.status.message['content']['text']
else:
    return '暂不支持此意图。'

主Agent根据任务状态选择不同的结果提取路径，这就是A2A协议状态机的实际消费方式。

---

五、子Agent内部：LangChain Agent + MCP工具

子Agent收到任务后，并不是直接执行SQL——它内部还有一个"小大脑"：LangChain Agent。这个Agent的作用是理解用户的自然语言，决定调用哪个MCP工具，传什么参数。

5.1 连接MCP服务获取工具

# a2a_server/a2a_weather.py — query_weather()
async def query_weather(conversation: str) -> dict:
    # 实例化MCP客户端，连接到天气MCP服务
    mcp_client = MultiServerMCPClient(MCP_WEATHER_SERVER_CFG)
    # 异步获取MCP服务提供的工具列表
    tools = await mcp_client.get_tools()

MultiServerMCPClient是LangChain提供的MCP适配器，它通过Streamable HTTP协议连接到MCP Server，自动发现并获取所有注册的工具。

MCP连接配置：

MCP_WEATHER_SERVER_CFG = {
    "weather_mcp": {
        "url": f"http://{CONFIG.MCP_WEATHER_HOST}:{CONFIG.MCP_WEATHER_PORT}/mcp",
        "transport": "streamable-http",
    }
}

注意URL末尾的/mcp后缀——这是MCP Streamable HTTP传输协议的标准端点。

5.2 构建LangChain Agent

    # 构建系统提示词
    agent_system_prompt = f"""你是专业天气查询助手，
    仅能调用提供的天气工具（query_weather 查历史/预报，query_weather_now 查实时），
    禁止编造天气数据。
    规则：
    1. 用户说"现在""当前""实时""今天"，或未指定日期时，调用 query_weather_now；
        用户说"明天""后天"或指定日期时，调用 query_weather；
    2. 用户说的明天、后天、未来N天，
        基于今日{today_str}转换为标准YYYY-MM-DD日期；
    3. 用户提问缺少城市，直接反问用户补齐，不能自行填充；
    4. 参数齐全后调用对应天气工具；
    5. 查到数据后用通顺中文整理。
    """

    # 创建LangChain Agent：LLM + MCP工具
    agent = create_agent(
        model=llm,
        tools=tools,
        system_prompt=agent_system_prompt
    )

    # 执行Agent
    agent_resp = await agent.ainvoke({
        "messages": [{"role": "user", "content": conversation}]
    })

create_agent创建的Agent具备自主推理能力——它会根据用户输入和系统提示词，自动决定调用哪个工具、传什么参数。这个过程对用户完全透明。

5.3 完整的调用链路

图2：子Agent内部完整调用链路——从A2A任务到天气数据的六步流转

---

六、MCP Server层：纯协议适配器

6.1 设计原则：零业务逻辑

MCP Server层的设计原则是纯协议适配，不包含任何业务逻辑。它只做三件事：

1. 创建FastMCP实例
2. 把mcp_tools/中的普通函数注册为MCP工具
3. 导出ASGI应用供uvicorn部署

# mcp_server/mcp_weather.py
from fastmcp import FastMCP
from mcp_tools.weather_tools import query_weather, query_weather_now

weather_mcp = FastMCP(
    name=CONFIG.MCP_WEATHER_NAME,
    instructions="天气查询工具，从MySQL weather_data表查询数据，支持实时天气查询",
)

# 把普通函数注册为MCP tool
weather_mcp.tool(
    name="query_weather",
    description="查询天气数据，参数：city(城市), start_date(开始日期), end_date(结束日期)",
    run_in_thread=False
)(query_weather)

weather_mcp.tool(
    name="query_weather_now",
    description="从和风天气API获取指定城市的实时天气，参数：city(城市名称)",
    run_in_thread=False
)(query_weather_now)

# 导出ASGI应用
_fastmcp_app = weather_mcp.http_app(transport='streamable-http')

6.2 三个MCP Server的工具清单

MCP Server	端口	工具数量	工具列表
mcp-weather	:5001	2	query_weather, query_weather_now
mcp-ticket	:5002	6	query_train, query_flight, query_concert, order_train, order_flight, order_concert
mcp-trip	:5003	6	query_car_rental, query_tour_group, query_insurance, order_car_rental, order_tour_group, order_insurance

共计14个MCP工具，覆盖天气/火车票/机票/演唱会/租车/旅游团/保险七大业务。

6.3 票务MCP Server注册示例

# mcp_server/mcp_ticket.py
from mcp_tools.ticket_tools import (
    query_train, query_flight, query_concert,
    order_train, order_flight, order_concert,
)

ticket_mcp = FastMCP(name=CONFIG.MCP_TICKET_NAME)

# 查询类
ticket_mcp.tool(name="query_train",
    description="查询火车票，参数：departure_city, arrival_city, date, seat_type(可选)"
)(query_train)

ticket_mcp.tool(name="query_flight",
    description="查询机票，参数：departure_city, arrival_city, date, cabin_type(可选)"
)(query_flight)

ticket_mcp.tool(name="query_concert",
    description="查询演唱会票，参数：city, artist, date, ticket_type(可选)"
)(query_concert)

# 预定类
ticket_mcp.tool(name="order_train",
    description="预定火车票，参数：departure_date, train_number, seat_type, number"
)(order_train)

ticket_mcp.tool(name="order_flight",
    description="预定机票，参数：departure_date, flight_number, cabin_type, number"
)(order_flight)

ticket_mcp.tool(name="order_concert",
    description="预定演出票，参数：start_date, artist, venue, ticket_type, number"
)(order_concert)

app = ticket_mcp.http_app(transport='streamable-http')

description参数至关重要——它会被LangChain Agent读取，用于决定在什么场景下调用哪个工具。描述写得越清晰，Agent的工具选择就越准确。

---

七、mcp_tools层：真正的业务逻辑

7.1 天气工具：定时采集 + 双通道查询

天气工具的实现分为两部分：

数据采集——WeatherCollector类定时从和风天气API采集6个城市（北京、上海、深圳、杭州、成都、郑州）的15天天气预报，存入MySQL的weather_data表：

# mcp_tools/weather_tools.py — WeatherCollector
class WeatherCollector:
    def __init__(self):
        self.cities = {
            "北京": "101010100", "上海": "101020100",
            "深圳": "101280601", "杭州": "101210101",
            "成都": "101270101", "郑州": "101180101"
        }
    
    def collect_and_save(self):
        """采集一个城市的天气数据并存入MySQL"""
        for city, location_id in self.cities.items():
            # 调用和风天气API获取15天预报
            forecast_data = self.fetch_forecast(location_id)
            # INSERT ... ON DUPLICATE KEY UPDATE（存在则更新，不存在则插入）
            self.save_weather(city, forecast_data)
    
    def start_scheduler(self):
        """定时任务：每1440分钟（1天）采集一次"""
        schedule.every(1440).minutes.do(self.collect_and_save)

查询接口——两个查询函数，一个查MySQL历史数据，一个调实时API：

def query_weather(city, start_date, end_date):
    """查询MySQL中的历史/预报天气数据"""
    sql = f"SELECT * FROM weather_data WHERE city='{city}' AND date BETWEEN '{start_date}' AND '{end_date}'"
    data = mysql.execute(sql)
    return json.dumps({"status": "success", "data": data})

def query_weather_now(city):
    """调用和风天气API获取实时天气"""
    # 先通过Geo API获取城市location_id
    location_id = get_location_id(city)
    # 调用实时天气API
    realtime_data = fetch_realtime_weather(location_id)
    return json.dumps({"status": "success", "data": realtime_data})

7.2 票务工具：SQL精确查询

票务工具是最标准的CRUD操作：

# mcp_tools/ticket_tools.py
def query_train(departure_city, arrival_city, date, seat_type=None):
    """查询火车票"""
    sql = f"""SELECT * FROM train_tickets 
              WHERE departure_city='{departure_city}' 
              AND arrival_city='{arrival_city}' 
              AND departure_date='{date}'"""
    if seat_type:
        sql += f" AND seat_type='{seat_type}'"
    data = mysql.execute(sql)
    return json.dumps({"status": "success", "data": data})

def order_train(departure_date, train_number, seat_type, number):
    """预定火车票（模拟）"""
    return json.dumps({
        "status": "success",
        "message": f"恭喜，预定成功！{train_number} {seat_type}x{number}"
    })

6个票务函数结构完全一致：3个query函数执行SELECT查询，3个order函数返回模拟的预定成功消息。

7.3 行程工具：RAG语义搜索的亮点

行程工具中最有技术含量的是query_tour_group()——它是整个项目中唯一使用RAG（检索增强生成）的函数。

传统SQL查询只能精确匹配关键词，但用户描述旅游需求时往往用自然语言，比如"想看雪山的短途旅行"。这时精确匹配就失效了，需要语义搜索。

# mcp_tools/trip_tools.py
def get_embedding(text):
    """调用阿里云Embedding API将文本转为1024维向量"""
    response = requests.post(
        "https://dashscope.aliyuncs.com/compatible-mode/v1/embeddings",
        headers={"Authorization": f"Bearer {CONFIG.EMBEDDING_API_KEY}"},
        json={"input": text, "model": "text-embedding-v3", "dimensions": 1024}
    )
    return response.json()["data"][0]["embedding"]

def search_tour_groups_in_milvus(query_text, city=None):
    """在Milvus向量库中语义搜索旅游团"""
    # 1. 将用户查询转为向量
    query_embedding = get_embedding(query_text)
    
    # 2. 在Milvus中做COSINE相似度搜索
    search_params = {"metric_type": "COSINE", "params": {"nprobe": 10}}
    results = milvus_collection.search(
        data=[query_embedding],
        anns_field="embedding",
        param=search_params,
        limit=5,
        expr=f"city == '{city}'" if city else None
    )
    return results

def query_tour_group(query_text, city=None):
    """语义搜索旅游团（RAG核心）"""
    results = search_tour_groups_in_milvus(query_text, city)
    # 格式化返回
    tour_groups = []
    for hits in results:
        for hit in hits:
            tour_groups.append({
                "tour_name": hit.entity.get("tour_name"),
                "description": hit.entity.get("description"),
                "price": hit.entity.get("price"),
                "similarity": hit.distance
            })
    return json.dumps({"status": "success", "data": tour_groups})

RAG流程可视化：

图3：旅游团语义搜索（RAG）流程——从自然语言到向量检索再到自然语言回复

举个例子：用户说"想看雪山的短途旅行"，传统SQL无法匹配任何记录，但通过向量相似度搜索，Milvus能找到描述中包含"雪山""冰川""高原"等语义相近的旅游团，即使团名中完全没有"雪山"二字。

---

八、配置驱动：新增Agent只需改YAML

SmartVoyage最具工程价值的设计之一是配置驱动的动态注册。新增一个Agent完全不需要修改主Agent代码：

8.1 config.yaml中的声明

# config.yaml
a2a_servers:
  weather:
    host: "127.0.0.1"
    port: 6001
    name: "WeatherQueryAssistant"
    prompt: |
      系统提示：您是一位专业的天气预报员...
      查询：{query}
      结果：{raw_response}
  ticket:
    host: "127.0.0.1"
    port: 6002
    name: "TicketAssistant"
    prompt: |
      系统提示：您是一位专业的旅行顾问...

mcp_servers:
  weather:
    host: "127.0.0.1"
    port: 5001
    name: "WeatherTools"
  ticket:
    host: "127.0.0.1"
    port: 5002
    name: "TicketTools"

8.2 Docker环境下的地址覆盖

本地开发用127.0.0.1，Docker部署用容器名（如a2a-weather）。SmartVoyage通过DockerConfig子类实现环境变量覆盖：

# utils/config.py
class DockerConfig(Config):
    def _apply_env_overrides(self):
        # 如果环境变量DOCKER_A2A_WEATHER_URL存在，则覆盖YAML中的host/port
        self._override_url_attrs("A2A_WEATHER", "a2a_weather")
        self._override_url_attrs("MCP_WEATHER", "mcp_weather")
        # ... 其他服务同理

CONFIG = DockerConfig.create_self()

Docker Compose中设置环境变量：

environment:
  - DOCKER_A2A_WEATHER_URL=http://a2a-weather:6001
  - DOCKER_MCP_WEATHER_URL=http://mcp-weather:5001

一套代码，两种环境——本地开发直连localhost，Docker部署自动切换到容器网络地址。

8.3 假设要新增一个"酒店Agent"

只需三步：

1. 在config.yaml中添加配置：

a2a_servers:
  hotel:
    host: "127.0.0.1"
    port: 6004
    name: "HotelAssistant"
    prompt: "..."

mcp_servers:
  hotel:
    host: "127.0.0.1"
    port: 5004
    name: "HotelTools"

1. 编写a2a_hotel.py和mcp_server/mcp_hotel.py（参照weather的模板）
2. 重启服务

主Agent的VoyageChatCore在启动时会自动发现新的AgentCard，将query hotel等技能纳入意图识别和路由映射。零代码修改，完全符合开闭原则。

---

九、A2A与MCP的关系：两层协议的分工

很多人会混淆A2A和MCP，这里明确它们各自的角色：

图4：A2A与MCP两层协议的分工——A2A管Agent间通信，MCP管工具调用

简单类比：

• A2A像是公司间的合同——规定"谁负责什么、怎么交付成果"
• MCP像是部门内的工具借用——规定"这个工具怎么用、需要什么参数"

子Agent通过A2A接收任务，通过MCP调用工具。两层协议各司其职，互不耦合。

---

十、部署方式：Waitress vs Uvicorn

三类服务的部署方式各有不同：

服务类型	框架	部署服务器	协议
A2A Server	Flask (python-a2a)	Waitress (WSGI)	HTTP
MCP Server	FastMCP (Starlette)	Uvicorn (ASGI)	Streamable HTTP
API网关	FastAPI	Uvicorn (ASGI)	HTTP/SSE

A2A Server使用Waitress而非Uvicorn，是因为python-a2a库基于Flask（WSGI），而MCP Server和API网关使用FastAPI/Starlette（ASGI），需要Uvicorn。

Docker中的启动命令：

# A2A Server (Waitress)
CMD ["waitress-serve", "--host=0.0.0.0", "--port=6001", "--threads=1", "a2a_weather:app"]

# MCP Server (Uvicorn)
CMD ["uvicorn", "mcp_weather:app", "--host", "0.0.0.0", "--port", "5001", "--workers", "2"]

---

十一、总结

本文从源码层面完整拆解了SmartVoyage的A2A协议层和MCP工具层，核心设计可以概括为：

AgentCard能力声明：每个子Agent通过AgentCard声明自己的技能列表，主Agent在启动时自动发现并构建路由映射。新增Agent无需修改主Agent代码。

A2A任务状态机：三种终态（Completed/InputRequired/Failed）覆盖了查询成功、参数不足追问、执行失败三种业务场景，主Agent根据状态选择结果提取路径。

LangChain Agent桥接：子Agent内部不是直接调SQL，而是通过LangChain Agent + MCP工具的组合，让LLM自主决定调用哪个工具、传什么参数。这赋予了子Agent处理自然语言的灵活性。

MCP纯协议适配：MCP Server层零业务逻辑，只负责把普通Python函数注册为MCP工具。业务逻辑完全下沉到mcp_tools层，职责清晰。

RAG语义搜索：旅游团查询使用Milvus向量库做COSINE相似度搜索，是项目中唯一的RAG实践，解决了传统SQL无法处理自然语言描述的痛点。

配置驱动 + 环境变量覆盖：config.yaml定义默认配置，DockerConfig通过环境变量覆盖，实现"一套代码、两种环境"。

下一篇文章将聚焦工程实践层面——双层缓存（Redis+Milvus）、四种记忆系统的实现细节、以及Docker 11容器的编排部署策略，敬请期待。

---