pip install fastmcp langchain-mcp-adapters

"""FastMCP 天气服务"""
from fastmcp import FastMCP
 
mcp = FastMCP("天气服务")
 
@mcp.tool()
def get_weather(city: str) -> str:
    """获取指定城市的天气信息
    Args:
        city: 城市名称，如 "北京"、"上海"、"广州"
    """
    # 模拟天气数据
    weather_data = {
        "北京": "晴天，气温 25°C，湿度 40%",
        "上海": "多云，气温 28°C，湿度 65%",
        "广州": "小雨，气温 30°C，湿度 80%",
        "深圳": "阴天，气温 29°C，湿度 75%",
    }
 
    if city in weather_data:
        return f"{city}天气：{weather_data[city]}"
    else:
        return f"{city}天气：晴，气温 22°C，湿度 50%"
 
mcp.run(transport="streamable-http", host="127.0.0.1", port=8000)
# mcp.run(transport="stdio")

"""LangChain MCP 客户端 - 连接 FastMCP 天气服务"""
import os
import asyncio
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool, BaseTool
from langchain_community.tools import WriteFileTool, ReadFileTool, ListDirectoryTool
from langchain_mcp_adapters.client import MultiServerMCPClient
 
load_dotenv()
prefix = "QWEN"
model = init_chat_model(
    model_provider="openai",
    configurable_fields=["model", "api_key", "base_url"],
    config_prefix=prefix
).with_config({
    "configurable": {
        f"{prefix}_model": os.getenv(f"{prefix}_MODEL"),
        f"{prefix}_api_key": os.getenv(f"{prefix}_API_KEY"),
        f"{prefix}_base_url": os.getenv(f"{prefix}_BASE_URL")
    }
})
 
class CalculateTool(BaseTool):
    name: str = "calculate"
    description: str = "计算数学表达式的值"
 
    def _run(self, expression: str) -> str:
        try:
            return f"计算结果: {eval(expression)}"
        except Exception as e:
            return f"计算错误: {str(e)}"
 
    async def _arun(self, expression: str) -> str:
        return self._run(expression)
 
async def main():
    client = MultiServerMCPClient(
        {"weather": {"url": "http://127.0.0.1:8000/mcp", "transport": "streamable_http"}}
        # stdio 模式：
        # {"weather": {"command": "python", "args": ["mcp_weather_stdio.py"]}}
    )
    mcp_tools = await client.get_tools()
 
    # ========== 初始化工具 ==========
    calculate = CalculateTool()
    write_file = WriteFileTool()
    read_file = ReadFileTool()
    list_dir = ListDirectoryTool()
 
    agent = create_agent(
        model=model,
        tools=[calculate, write_file, read_file, list_dir] + mcp_tools,
        system_prompt="你是一个助手，会用工具计算、读写文件、列出目录、查询天气。",
        debug=True
    )
 
    queries = [
        "北京天气怎么样？",
        "计算 2024*12+500，然后把结果保存到 result.txt",
        "读取 result.txt 的内容",
    ]
 
    for q in queries:
        print(f"\n问：{q}")
        result = await agent.ainvoke({"messages": [{"role": "user", "content": q}]})
        print(f"答：{result['messages'][-1].content}")
 
asyncio.run(main())