MCP Weather Agent

这是一个使用模型上下文协议（MCP）与LangGraph进行智能工具编排的生产就绪且可扩展的示例。
该项目展示了干净架构原则，并作为构建基于代理系统的一个蓝图，这些系统通过MCP动态集成工具。

项目结构

mcp-weather-agent/
├── src/
│   ├── __init__.py
│   ├── main.py                          # 入口点
│   ├── config.py                        # 集中配置
│   ├── graph/                           # LangGraph工作流
│   │   ├── __init__.py
│   │   ├── state.py                     # 代理状态模式
│   │   ├── graph_builder.py             # 图组装
│   │   ├── nodes/                       # 图节点
│   │   │   ├── __init__.py
│   │   │   ├── agent.py                 # 代理节点（模型调用）
│   │   │   └── tools.py                 # 工具执行节点
│   │   └── edges/                       # 条件路由
│   │       ├── __init__.py
│   │       └── agent_to_tools.py        # 代理→工具决策逻辑
│   ├── tools/                           # 外部工具
│   │   ├── __init__.py
│   │   └── weather_mcp/
│   │       ├── __init__.py
│   │       ├── client.py                # MCP客户端初始化
│   │       └── server.py                # 带有天气工具的MCP服务器
│   └── utils/                           # 实用工具
│       ├── __init__.py
│       └── logger.py                    # 集中日志记录
├── requirements.txt
├── .env.example
├── .gitignore
└── README.md

工具

工具	目的	返回值
get_current_weather(location)	温度、风速、天气状况	温度、体感温度、风速/风向
get_atmospheric_conditions(location)	空气属性	湿度、气压、云量、能见度、降水量、紫外线指数
get_astronomical_data(location)	日月数据	日出、日落、月升、月落、月相
get_air_quality(location)	污染水平	CO、NO₂、O₃、SO₂、PM2.5、PM10、空气质量指数

设置

1. 安装依赖

pip install -r requirements.txt

2. 获取API密钥

• 天气：从weatherapi.com获取免费密钥
• LLM：从openrouter.ai获取免费层级密钥

3. 配置

复制.env.example并添加你的API密钥：

cp .env.example .env

然后编辑.env文件，添加你的密钥：

WEATHER_API_KEY=your_key_here
MODEL_API_KEY=your_key_here

4. 运行

python -m src.main

工作原理

执行流程

1. 初始化MCP客户端
   └─ 从MCP服务器加载天气工具

2. 构建LangGraph
   ├─ 创建代理节点（LLM + 工具）
   ├─ 创建工具节点（工具执行）
   └─ 创建代理→工具边（路由）

3. 执行图
   ├─ 代理处理用户查询
   ├─ 代理决定调用哪些工具
   ├─ 工具节点执行选定工具（并发）
   ├─ 结果返回给代理
   └─ 代理生成最终响应

此设计将MCP客户端、工具和图逻辑解耦，实现可扩展性和可扩展性，并具有清晰的边界。

异步执行

工具执行节点使用async/await并发运行工具：

# src/graph/nodes/tools.py
async def process_tool_calls(state: AgentState) -> AgentState:
    """并行执行多个工具。"""
    # 如果代理调用了多个工具，则它们会并发运行
    # 示例：compare_air_quality() 调用 get_air_quality(Tehran) + get_air_quality(NewYork)
    # 两个API调用同时发生，而不是顺序发生

注意：系统默认并发执行工具。然而，如果工具之间存在依赖关系或顺序重要，你可以修改工具执行逻辑，在需要时按顺序运行它们。

可扩展图结构

节点 – 状态转换器

• agent.py – 注册工具的LLM调用
• tools.py – 工具执行引擎
• 根据需要添加更多：validation.py，formatting.py等。

边 – 路由逻辑

• agent_to_tools.py – 决定是否需要工具
• 易于扩展：tools_to_validator.py，validator_to_formatter.py等。

为什么这很重要

• 每个节点/边都有一个单一职责
• 新节点可以添加而无需重构
• 保证数据通过LangGraph代理的清晰流动

代理状态模式（graph/state.py）

状态是流经图的数据。它使用TypedDict定义以确保类型安全：

from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    """天气代理工作流的状态。"""
    messages: list[BaseMessage]

如何工作：

• messages – 对话消息列表（用户查询、代理响应、工具结果）
• 追加模式 – 每个节点在状态中添加新消息，而不是替换它
• 清晰历史 – 所有消息都被保存，使对话流程透明且易于调试

示例状态演变：

步骤1（初始）：
  messages = [HumanMessage("比较阿姆斯特丹和纽约的空气质量")]

步骤2（代理节点后）：
  messages = [HumanMessage(...), AIMessage(tool_calls=[call_1, call_2])]

步骤3（工具节点后）：
  messages = [HumanMessage(...), AIMessage(...), ToolMessage("结果1"), ToolMessage("结果2")]

步骤4（最终）：
  messages = [HumanMessage(...), AIMessage(...), ToolMessage(...), ToolMessage(...), AIMessage("最终答案")]

扩展状态：

随着代理的增长，你可以添加更多字段：

class AdvancedAgentState(TypedDict):
    messages: list[BaseMessage]
    user_id: str                    # 跟踪哪个用户发起了请求
    metadata: dict                  # 存储查询元数据
    tool_calls_count: int           # 监控工具使用情况

配置系统（config.py）

所有应用程序设置的单一真实来源：

• 通过环境变量轻松覆盖
• 使用dataclasses确保类型安全

@dataclass
class ModelConfig:
    model_name: str
    base_url: Optional[str]
    api_key: Optional[str]
    temperature: float
    max_tokens: int

为什么：创建时验证，从代码中易于理解，确保运行时行为可预测。

统一日志系统（utils/logger.py）

from src.utils.logger import get_logger

logger = get_logger(__name__)  # 在任何地方使用

• 集中式日志配置
• 日志级别通过LOG_LEVEL环境变量控制
• 整个应用中的日志格式一致
• 易于扩展（文件处理器、云日志等）

设计模式使用

1. 工厂模式（节点）

def create_agent_node(model_with_tools):
    def agent_node(state):
        # 节点逻辑
        pass
    return agent_node

为什么：将模型上下文保留在节点内，避免全局变量和重复设置。

2. 配置即代码

类型安全的配置对象使运行时设置可预测且可扩展。

@dataclass
class ModelConfig:
    model_name: str
    base_url: Optional[str]
    api_key: Optional[str]
    temperature: float
    max_tokens: int

为什么：配置明确且经过验证。

3. 依赖注入

工具动态传递给build_graph()而不是硬编码。

为什么：提高测试性、灵活性和关注分离。

扩展代理

添加新节点

1. 创建src/graph/nodes/your_node.py
2. 实现节点函数
3. 在graph_builder.py中注册它

示例：

# src/graph/nodes/formatter.py
def create_formatter_node():
    def formatter_node(state: AgentState) -> AgentState:
        # 格式化响应
        return {"messages": state["messages"] + [formatted]}
    return formatter_node

添加新边

1. 创建src/graph/edges/your_edge.py
2. 实现路由函数
3. 导出到src/graph/edges/__init__.py
4. 添加到graph_builder.py

示例：

# src/graph/edges/tools_to_formatter.py
def should_format(state: AgentState) -> str:
    # 条件路由
    return "formatter" if condition else END

添加新MCP工具

1. 在src/tools/weather_mcp/server.py中添加工具函数
2. 使用@mcp.tool()装饰
3. 包含清晰详细的文档字符串
4. 工具在启动时自动注册到MCP客户端

要扩展到天气以外的领域，只需添加新的工具包：

• src/tools/finance_mcp/
• src/tools/drug_mcp/

每个MCP工具集都保持独立和便携。

添加新实用工具

在src/utils/中添加新文件用于共享逻辑：

• src/utils/validators.py → 数据验证
• src/utils/formatters.py → 响应格式化

这些实用工具可以在任何地方导入，以维护一致的基础架构层。

关键见解：设计MCP服务器

1. 创建专注的单用途工具

将功能拆分为专门的功能，而不是一个庞大的工具：

差的设计：

@mcp.tool()
def get_all_weather(location):
    # 返回所有内容：温度、空气质量、天文数据...
    # 代理无法区分什么相关

好的设计：

@mcp.tool()
def get_current_weather(location):
    # 仅温度、风速和天气状况

@mcp.tool()
def get_current_air_quality(location):
    # 仅污染数据

为什么：代理智能地选择所需的内容，引导交互，导致更清晰的语义和减少推理复杂性。

2. 编写清晰详细的描述

全面的文档字符串使LLM能够理解何时以及如何使用每个工具：

好的描述：

@mcp.tool()
def get_current_air_quality(location: str) -> dict:
    """
    获取给定位置的当前空气质量数据。

    参数：
        location: 城市名称或位置（例如，“纽约”，“伦敦”，“德黑兰”）

    返回：
        包含以下内容的字典：
        - CO（一氧化碳）水平
        - NO2（二氧化氮）水平
        - O3（臭氧）水平
        - SO2（二氧化硫）水平
        - PM2.5（细颗粒物）水平
        - PM1.0（粗颗粒物）水平
        - 美国环保署空气质量指数
        - 英国DEFRA空气质量指数
    """

差的描述：

@mcp.tool()
def get_current_air_quality(location: str) -> dict:
    """获取空气质量信息。"""  # 没有关于返回什么数据的细节！

为什么：LLM依赖描述来做出工具选择决策。更好的描述 = 更聪明的路由。

自动工具选择实例

有了专注的工具和清晰的描述，代理自动正确路由查询：

• "伦敦的天气怎么样？" → get_current_weather(location="伦敦")
• "德黑兰的空气质量好吗？" → get_current_air_quality(location="德黑兰")
• "比较德黑兰和纽约的空气质量" → get_current_air_quality(location="德黑兰") + get_current_air_quality(location="纽约")
• "告诉我巴黎的天气和空气质量" → get_current_weather(location="巴黎") + get_current_air_quality(location="巴黎")

不需要硬编码逻辑——代理根据你的工具描述自行解决。

扩展此蓝图

适应此架构以应用于自己的领域：

• 将weather_mcp替换为自定义MCP工具
• 添加特定领域的节点或边
• 修改config.py中的配置
• 重用logger和graph_builder模式

此架构干净地扩展，提供了一个清晰、可扩展的基础，适用于多工具、特定领域的代理。

不使用MCP使用此架构

虽然本项目使用了MCP（模型上下文协议），但核心架构在没有MCP的情况下也能完美运行。以下是调整方法：

步骤1：替换MCP客户端

代替src/tools/weather_mcp/client.py，创建直接工具实现：

# src/tools/direct_tools.py
from langchain_core.tools import tool

@tool
def get_current_weather(location: str) -> dict:
    """获取某个地点的当前天气。"""
    # 直接API调用或本地逻辑
    response = requests.get(f"https://api.weatherapi.com/v1/current.json", ...)
    return response.json()

@tool
def get_air_quality(location: str) -> dict:
    """获取某个地点的空气质量数据。"""
    # 你的实现
    pass

# 创建工具列表
tools = [get_current_weather, get_air_quality]

步骤2：更新main.py

替换MCP初始化为直接工具加载：

# src/main.py（无MCP）
async def main():
    logger.info("正在加载工具...")
    from src.tools.direct_tools import tools  # 直接导入而不是MCP

    logger.info("正在构建代理图...")
    compiled_graph = build_graph(tools)

    # 其余部分保持不变...

步骤3：图构建器保持不变

graph_builder.py不需要更改，因为它接受来自任何来源的工具（MCP、REST或内联）：

# src/graph/graph_builder.py - 适用于MCP和直接工具
def build_graph(tools):  # 工具可以来自任何地方
    model = ChatOpenAI(...)
    model_with_tools = model.bind_tools(tools)  # 无论工具来源如何都能工作
    # ... 图构建的其余部分

MCP服务器/客户端生命周期

组件	运行于	管理方式	目的 / 责任
MCP客户端	代理主进程	MultiServerMCPClient	发现并加载可用的MCP工具
MCP服务器	单独子进程	MCP框架（启动）	执行工具逻辑，返回结果

此生命周期确保您的代理具有容错能力——如果一个服务器进程失败，客户端可以重新连接或选择性重试，而不会中断主要代理流程。

对比：MCP vs 直接工具

方面	MCP	直接工具
故障隔离	工具崩溃不影响代理	工具崩溃会导致整个代理崩溃
语言支持	多语言工具（JS、Rust、Python等）	Python仅限
性能	稍微的进程间通信开销	更快（无进程间通信开销）
设置与调试	更复杂（单独进程）	更简单（单进程，更容易调试）
最佳适用场景	生产系统，分布式工具，容错	原型设计，简单项目，仅限Python的堆栈