MCP气象员

A. 生产就绪，可扩展 使用示例 模型上下文协议（MCP） 随着 LangGraph 用于智能工具编排。\ 该项目展示了 清洁建筑原则 并作为蓝图用于构建基于代理的系统，通过MCP动态集成工具。

项目结构

mcp-weather-agent/
├── src/
│   ├── __init__.py
│   ├── main.py                          # Entry point
│   ├── config.py                        # Centralized configuration
│   ├── graph/                           # LangGraph workflow
│   │   ├── __init__.py
│   │   ├── state.py                     # Agent state schema
│   │   ├── graph_builder.py             # Graph assembly
│   │   ├── nodes/                       # Graph nodes
│   │   │   ├── __init__.py
│   │   │   ├── agent.py                 # Agent node (model invocation)
│   │   │   └── tools.py                 # Tool execution node
│   │   └── edges/                       # Conditional routing
│   │       ├── __init__.py
│   │       └── agent_to_tools.py        # Agent → Tools decision logic
│   ├── tools/                           # External tools
│   │   ├── __init__.py
│   │   └── weather_mcp/
│   │       ├── __init__.py
│   │       ├── client.py                # MCP client initialization
│   │       └── server.py                # MCP server with weather tools
│   └── utils/                           # Utilities
│       ├── __init__.py
│       └── logger.py                    # Centralized logging
├── 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. Initialize MCP Client
   └─ Load weather tools from MCP server

2. Build LangGraph
   ├─ Create Agent Node (LLM + tools)
   ├─ Create Tool Node (tool execution)
   └─ Create Agent→Tools Edge (routing)

3. Execute Graph
   ├─ Agent processes user query
   ├─ Agent decides which tools to call
   ├─ Tool Node executes selected tools (concurrently)
   ├─ Results returned to Agent
   └─ Agent generates final response

这种设计将 MCP客户端, 工具，以及 图形逻辑，以明确的边界实现可扩展性和可扩展性。

异步执行

工具执行节点运行工具同时使用async/await：

# src/graph/nodes/tools.py
async def process_tool_calls(state: AgentState) -> AgentState:
    """Execute multiple tools in parallel."""
    # If the agent calls multiple tools, they run concurrently
    # Example: compare_air_quality() calls get_air_quality(Tehran) + get_air_quality(NewYork)
    # Both API calls happen simultaneously, not sequentially

注：默认情况下，系统同时执行工具。但是，如果工具有依赖关系或顺序问题，您可以修改工具执行逻辑，以便在需要时按顺序运行它们。

可扩展图结构

节点——状态转换器

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):
    """State for the weather agent workflow."""
    messages: list[BaseMessage]

工作原理：

messages –对话消息列表（用户查询、代理响应、工具结果）
仅添加图案 –每个节点都会向状态中添加新消息，而不是替换它
清除历史记录 –保留所有消息，使对话流程透明且可调试

状态演变示例：

Step 1 (Initial):
  messages = [HumanMessage("Compare air quality between Amsterdam and New York")]

Step 2 (After Agent Node):
  messages = [HumanMessage(...), AIMessage(tool_calls=[call_1, call_2])]

Step 3 (After Tool Node):
  messages = [HumanMessage(...), AIMessage(...), ToolMessage("Result 1"), ToolMessage("Result 2")]

Step 4 (Final):
  messages = [HumanMessage(...), AIMessage(...), ToolMessage(...), ToolMessage(...), AIMessage("Final Answer")]

扩展国家：

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

class AdvancedAgentState(TypedDict):
    messages: list[BaseMessage]
    user_id: str                    # Track which user made the request
    metadata: dict                  # Store query metadata
    tool_calls_count: int           # Monitor tool usage

配置系统(`config.py`)

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

易于通过环境变量进行覆盖
使用数据类进行类型安全

@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__)  # Used everywhere

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

使用的设计模式

1. 工厂模式（节点）

def create_agent_node(model_with_tools):
    def agent_node(state):
        # Node logic
        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() 而不是硬编码。

为什么？ 提高了可测试性、灵活性和关注点分离。

扩展代理

添加新节点

创建 src/graph/nodes/your_node.py
实现节点功能
在中注册 graph_builder.py

例子：

# src/graph/nodes/formatter.py
def create_formatter_node():
    def formatter_node(state: AgentState) -> AgentState:
        # Format the response
        return {"messages": state["messages"] + [formatted]}
    return formatter_node

添加新边

创建 src/graph/edges/your_edge.py
实现路由功能
出口 src/graph/edges/__init__.py
添加到 graph_builder.py

例子：

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

添加新的MCP工具

在中添加工具功能 src/tools/weather_mcp/server.py
用…装饰 @mcp.tool()
包含清晰、详细的文档字符串
工具在启动时自动向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):
    # Returns everything: temperature, air quality, astronomy...
    # Agent can't distinguish what's relevant

好的设计：

@mcp.tool()
def get_current_weather(location):
    # Only temperature, wind, and weather conditions

@mcp.tool()
def get_current_air_quality(location):
    # Only pollution data

为什么？ 代理智能地只选择它需要的东西，实现引导式交互，从而获得更清晰的语义并降低推理复杂性。

2. 写清楚、详细的描述

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

良好描述：

@mcp.tool()
def get_current_air_quality(location: str) -> dict:
    """
    Get current air quality data for a given location.

    Args:
        location: The city name or location (e.g., "New York", "London", "Tehran")

    Returns:
        A dictionary containing:
        - CO (Carbon Monoxide) levels
        - NO2 (Nitrogen Dioxide) levels
        - O3 (Ozone) levels
        - SO2 (Sulfur Dioxide) levels
        - PM2.5 (Fine Particulate Matter) levels
        - PM10 (Coarse Particulate Matter) levels
        - EPA Air Quality Index
        - GB DEFRA Air Quality Index
    """

糟糕的描述：

@mcp.tool()
def get_current_air_quality(location: str) -> dict:
    """Get air quality information."""  # No details about what data is returned!

为什么？ LLM依靠描述来做出工具选择决策。更好的描述=更智能的路由。

自主工具选择在行动中

通过集中的工具和清晰的描述，代理可以自动正确地路由查询：

"What's the weather in London?" → get_current_weather(location="London")
"Is the air quality good in Tehran?" → get_current_air_quality(location="Tehran")
"Compare air quality between Tehran and New York" → get_current_air_quality(location="Tehran") + get_current_air_quality(location="New York")
"Tell me about weather and air quality in Paris" → get_current_weather(location="Paris") + get_current_air_quality(location="Paris")

不需要硬编码的逻辑——代理会从您的工具描述中找出它。

扩展此蓝图

将此架构适应您自己的域：

替换 weather_mcp 使用自定义MCP工具
添加特定于域的节点或边
修改配置 config.py
复用 logger 和 graph_builder 模式

此架构 鳞片干净 并为多工具、特定领域的代理提供了清晰、可扩展的基础。

在没有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:
    """Get current weather for a location."""
    # Direct API call or local logic
    response = requests.get(f"https://api.weatherapi.com/v1/current.json", ...)
    return response.json()

@tool
def get_air_quality(location: str) -> dict:
    """Get air quality data for a location."""
    # Your implementation
    pass

# Create tool list
tools = [get_current_weather, get_air_quality]

步骤2：更新Main.py

用直接工具加载代替MCP初始化：

# src/main.py (without MCP)
async def main():
    logger.info("Loading tools...")
    from src.tools.direct_tools import tools  # Direct import instead of MCP

    logger.info("Building agent graph...")
    compiled_graph = build_graph(tools)

    # Rest remains the same...

步骤3：图形生成器按原样工作

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

# src/graph/graph_builder.py - Works with both MCP and direct tools
def build_graph(tools):  # Tools can come from anywhere
    model = ChatOpenAI(...)
    model_with_tools = model.bind_tools(tools)  # Works regardless of tool source
    # ... rest of graph building

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

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

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

比较：MCP与直接工具

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

MCP Weather Agent

快速接入

详细介绍