跳到主要内容
极客日志极客日志面向AI+效率的开发者社区
首页博客GitHub 精选镜像工具UI配色美学隐私政策关于联系
搜索内容 / 工具 / 仓库 / 镜像...⌘K搜索
注册
博客列表
PythonAI算法

基于 Claude MCP 协议的智能体落地示例

综述由AI生成基于 Model Context Protocol (MCP) 构建智能体,解决大模型内容非实时性问题。通过构建提供天气警报和预报工具的 MCP 服务器,并连接至 Claude for Desktop 客户端,演示了完整的落地链路。文章涵盖环境配置、服务端工具注册与实现、客户端交互逻辑及配置文件设置。MCP 协议使得智能体能够自动发现和管理工具,简化了多 API 集成时的 Prompt 构建与维护工作,提升了大模型与外部数据交互的能力。

JavaCoder发布于 2026/2/8更新于 2026/5/262.1K 浏览
基于 Claude MCP 协议的智能体落地示例

1. 背景

在《MCP(Model Context Protocol) 大模型智能体第一个开源标准协议》一文中,介绍了 MCP 的概念、架构及解决的问题,但缺少具体示例来理解整套框架如何落地。

本文基于 Claude 官方示例——获取天气预报,演示 MCP 落地的整条链路。

2. MCP 示例

该案例构建一个简单的 MCP 天气预报服务器,并将其连接到主机(Claude for Desktop)。大模型虽然能力强,但内容具有非实时性,无法直接获取最新天气。通过 MCP 可以解决这一问题。

构建一个提供两个工具的服务器:获取警报(get-alerts)和获取预报(get-forecast),然后连接至 MCP 主机。

环境配置

(1)安装 uv

curl -LsSf https://astral.sh/uv/install.sh | sh 

安装完成后提示:

downloading uv 0.6.9 aarch64-apple-darwin no checksums to verify installing to /Users/nicolas/.local/bin uv uvx everything's installed!

(2)安装所需的依赖包

文章配图

(3)在 server.py 中构建相应的 get-alerts 和 get-forecast 工具:

from typing import Any
import asyncio
import httpx
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio

NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

server = Server("weather")

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """ List available tools."""
    return [
        types.Tool(
            name="get-alerts",
            description="Get weather alerts for a state",
            inputSchema={
                : ,
                : {
                    : {: , : },
                },
                : [],
            },
        ),
        types.Tool(
            name=,
            description=,
            inputSchema={
                : ,
                : {
                    : {: , : },
                    : {: , : },
                },
                : [, ],
            },
        ),
    ]

  () -> [, ] | :
    headers = {: USER_AGENT, : }
    :
        response =  client.get(url, headers=headers, timeout=)
        response.raise_for_status()
         response.json()
     Exception:
         

 () -> :
    props = feature[]
     (
        
        
        
        
        
        
    )


  () -> [types.TextContent | types.ImageContent | types.EmbeddedResource]:
      arguments:
         ValueError()
    
     name == :
        state = arguments.get()
          state:
             ValueError()
        state = state.upper()
         (state) != :
             ValueError()
          httpx.AsyncClient()  client:
            alerts_url = 
            alerts_data =  make_nws_request(client, alerts_url)
              alerts_data:
                 [types.TextContent(=, text=)]
            features = alerts_data.get(, [])
              features:
                 [types.TextContent(=, text=)]
            formatted_alerts = [format_alert(feature)  feature  features[:]]
            alerts_text =  + .join(formatted_alerts)
             [types.TextContent(=, text=alerts_text)]
    
     name == :
        :
            latitude = (arguments.get())
            longitude = (arguments.get())
         (TypeError, ValueError):
             [types.TextContent(=, text=)]
        
          (- <= latitude <= )   (- <= longitude <= ):
             [types.TextContent(=, text=)]
        
          httpx.AsyncClient()  client:
            points_url = 
            points_data =  make_nws_request(client, points_url)
              points_data:
                 [types.TextContent(=, text=)]
            
            properties = points_data.get(, {})
            forecast_url = properties.get()
              forecast_url:
                 [types.TextContent(=, text=)]
            
            forecast_data =  make_nws_request(client, forecast_url)
              forecast_data:
                 [types.TextContent(=, text=)]
            
            periods = forecast_data.get(, {}).get(, [])
              periods:
                 [types.TextContent(=, text=)]
            
            formatted_forecast = []
             period  periods:
                forecast_text = (
                    
                    
                    
                    
                    
                )
                formatted_forecast.append(forecast_text)
            
            forecast_text =  + .join(formatted_forecast)
             [types.TextContent(=, text=forecast_text)]
    :
         ValueError()

  ():
      mcp.server.stdio.stdio_server()  (read_stream, write_stream):
         server.run(
            read_stream, write_stream,
            InitializationOptions(
                server_name=,
                server_version=,
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )

 __name__ == :
    asyncio.run(main())
"type"
"object"
"properties"
"state"
"type"
"string"
"description"
"Two-letter state code (e.g. CA, NY)"
"required"
"state"
"get-forecast"
"Get weather forecast for a location"
"type"
"object"
"properties"
"latitude"
"type"
"number"
"description"
"Latitude of the location"
"longitude"
"type"
"number"
"description"
"Longitude of the location"
"required"
"latitude"
"longitude"
async
def
make_nws_request
client: httpx.AsyncClient, url: str
dict
str
Any
None
"User-Agent"
"Accept"
"application/geo+json"
try
await
30.0
return
except
return
None
def
format_alert
feature: dict
str
"properties"
return
f"Event: {props.get('event', 'Unknown')}\n"
f"Area: {props.get('areaDesc', 'Unknown')}\n"
f"Severity: {props.get('severity', 'Unknown')}\n"
f"Status: {props.get('status', 'Unknown')}\n"
f"Headline: {props.get('headline', 'No headline')}\n"
"---"
@server.call_tool()
async
def
handle_call_tool
name: str, arguments: dict | None
list
if
not
raise
"Missing arguments"
if
"get-alerts"
"state"
if
not
raise
"Missing state parameter"
if
len
2
raise
"State must be a two-letter code (e.g. CA, NY)"
async
with
as
f"{NWS_API_BASE}/alerts?area={state}"
await
if
not
return
type
"text"
"Failed to retrieve alerts data"
"features"
if
not
return
type
"text"
f"No active alerts for {state}"
for
in
20
f"Active alerts for {state}:\n\n"
"\n"
return
type
"text"
elif
"get-forecast"
try
float
"latitude"
float
"longitude"
except
return
type
"text"
"Invalid coordinates."
if
not
90
90
or
not
180
180
return
type
"text"
"Invalid coordinates."
async
with
as
f"{NWS_API_BASE}/points/{latitude},{longitude}"
await
if
not
return
type
"text"
f"Failed to retrieve grid point data."
"properties"
"forecast"
if
not
return
type
"text"
"Failed to get forecast URL."
await
if
not
return
type
"text"
"Failed to retrieve forecast data"
"properties"
"periods"
if
not
return
type
"text"
"No forecast periods available"
for
in
f"{period.get('name', 'Unknown')}:\n"
f"Temperature: {period.get('temperature', 'Unknown')}°{period.get('temperatureUnit', 'F')}\n"
f"Wind: {period.get('windSpeed', 'Unknown')} {period.get('windDirection', '')}\n"
f"{period.get('shortForecast', 'No forecast available')}\n"
"---"
f"Forecast for {latitude}, {longitude}:\n\n"
"\n"
return
type
"text"
else
raise
f"Unknown tool: {name}"
async
def
main
async
with
as
await
"weather"
"0.1.0"
if
"__main__"

核心在于 @server.list_tools() 注册可用工具处理器,以及 @server.call_tool() 注册执行工具调用的处理器。

调用逻辑匹配工具名称,抽取输入参数,发起 API 请求并处理结果。

文章配图

服务端与客户端交互

测试服务器与 Claude for Desktop。构建 MCP 客户端的核心逻辑如下:

async def process_query(self, query: str) -> str:
    messages = [{"role": "user", "content": query}]
    response = await self.session.list_tools()
    available_tools = [{"name": tool.name, "description": tool.description, "input_schema": tool.inputSchema} for tool in response.tools]
    
    response = self.anthropic.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1000,
        messages=messages,
        tools=available_tools
    )
    
    final_text = []
    assistant_message_content = []
    for content in response.content:
        if content.type == 'text':
            final_text.append(content.text)
            assistant_message_content.append(content)
        elif content.type == 'tool_use':
            tool_name = content.name
            tool_args = content.input
            result = await self.session.call_tool(tool_name, tool_args)
            final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")
            assistant_message_content.append(content)
            messages.append({"role": "assistant", "content": assistant_message_content})
            messages.append({"role": "user", "content": [{"type": "tool_result", "tool_use_id": content.id, "content": result.content}]})
            
            response = self.anthropic.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1000,
                messages=messages,
                tools=available_tools
            )
            final_text.append(response.content[0].text)
    return "\n".join(final_text)

启动客户端需打开配置文件:~/Library/Application Support/Claude/claude_desktop_config.json。若不存在则创建,配置如下:

{
  "mcpServers": {
    "weather": {
      "command": "uv",
      "args": ["--directory", "/ABSOLUTE/PATH/TO/PARENT/FOLDER/weather", "run", "server.py"]
    }
  }
}

保存并重启 Claude for Desktop,即可识别暴露的工具。

文章配图

文章配图

询问天气时,会提示调用 get-forecast 的 tool。

文章配图

3. MCP 到底解决了什么问题

工具是智能体框架的重要组成部分。即使没有 MCP 协议也能实现 LLM 智能体,但存在弊端:当有许多不同 API 时,启用工具使用很麻烦,因为任何工具都需要手动构建 prompt,且每当其 API 发生变化时需手动更新。

MCP 解决了当存在大量工具时,能够自动发现并自动构建 prompt 的问题。

整体流程示例:

(1)以总结 git 项目最近 5 次提交为例,MCP 主机将首先调用 MCP 服务器,询问有哪些工具可用。

文章配图

(2)MCP 客户端接收到所列出的可用工具后发给 LLM,LLM 选择使用某个工具,通过主机向 MCP 服务器发送请求,接收结果。

文章配图

(3)LLM 收到工具处理结果后,向用户输出最终答案。

文章配图

总结起来,MCP 协议让智能体更容易管理、发现、使用工具。

4. 参考资料

【1】For Server Developers - Model Context Protocol 【2】For Client Developers - Model Context Protocol 【4】MCP 工作原理

目录

  1. 1. 背景
  2. 2. MCP 示例
  3. 环境配置
  4. 服务端与客户端交互
  5. 3. MCP 到底解决了什么问题
  6. 4. 参考资料
  • 💰 8折买阿里云服务器限时8折了解详情
  • Magick API 一键接入全球大模型注册送1000万token查看
  • 🤖 一键搭建Deepseek满血版了解详情
  • 一键打造专属AI 智能体了解详情
极客日志微信公众号二维码

微信扫一扫,关注极客日志

微信公众号「极客日志V2」,在微信中扫描左侧二维码关注。展示文案:极客日志V2 zeeklog

更多推荐文章

查看全部
  • C语言数据结构与算法基础:文件操作、排序查找及链表简介
  • macOS HBuilderX 运行 uniapp 报错缺少 Node.js 环境解决方案
  • Android Studio 安装及 SDK、JDK、Gradle 配置指南
  • Open WebUI MCPo 项目解析:将 MCP 工具转换为 OpenAPI 接口
  • 危险区域闯入提醒系统基于 GLM-4.6V-Flash-WEB 的实现
  • Flutter 三方库 webkit_inspection_protocol 的 OpenHarmony 适配指南
  • 单链表实战:删除指定节点、反转链表与查找中间节点
  • 单链表高频题解:删除节点、反转链表与查找中间节点
  • 单链表综合练习:删除指定节点、反转链表与查找中间节点
  • 基于 SSM 框架的校园二手物品交易网站设计与实现
  • 基于 SpringBoot 的体育竞技门票售卖系统设计与实现
  • 昇腾 910B NPU 平台 ops-transformer 算子性能测试与 PyTorch 对比
  • 数据结构:栈与队列的应用及矩阵压缩存储
  • ROS 2 实战:海龟仿真器启动与 ros2 run 命令详解
  • 前缀和与二维前缀和算法模板详解
  • Web Components 核心原理与实战指南
  • JavaScript 与 TypeScript 的本质区别、优缺点及语法差异详解
  • Ling Studio 一站式 AI 生产力平台功能与实战指南
  • ROS 2 实战:海龟仿真器与 ros2 run 命令详解
  • 二叉树 DFS 实战:计算布尔值与路径数字和
  • 相关免费在线工具

    • 加密/解密文本

      使用加密算法(如AES、TripleDES、Rabbit或RC4)加密和解密文本明文。 在线工具,加密/解密文本在线工具,online

    • RSA密钥对生成器

      生成新的随机RSA私钥和公钥pem证书。 在线工具,RSA密钥对生成器在线工具,online

    • Mermaid 预览与可视化编辑

      基于 Mermaid.js 实时预览流程图、时序图等图表,支持源码编辑与即时渲染。 在线工具,Mermaid 预览与可视化编辑在线工具,online

    • 随机西班牙地址生成器

      随机生成西班牙地址(支持马德里、加泰罗尼亚、安达卢西亚、瓦伦西亚筛选),支持数量快捷选择、显示全部与下载。 在线工具,随机西班牙地址生成器在线工具,online

    • Gemini 图片去水印

      基于开源反向 Alpha 混合算法去除 Gemini/Nano Banana 图片水印,支持批量处理与下载。 在线工具,Gemini 图片去水印在线工具,online

    • curl 转代码

      解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online