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

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

基于 Claude MCP 协议的智能体落地示例介绍了 Model Context Protocol 的概念及架构,通过构建天气预报服务器连接 Claude for Desktop 的完整链路,演示了 MCP 的实际落地流程。内容涵盖环境配置、工具注册、服务端与客户端交互逻辑,以及 MCP 如何解决大模型工具调用繁琐的问题,实现工具的自动发现与 Prompt 构建。

不知所云发布于 2026/2/7更新于 2026/10/24101 浏览
基于 Claude MCP 协议的智能体落地示例

1. 背景

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

今天我们基于 Claude 的官方例子——获取天气预报,来理解 MCP 落地的整条链路。

2. MCP 示例

该案例是构建一个简单的 MCP 天气预报服务器,并将其连接到主机,即 Claude for Desktop。从基本设置开始,然后逐步发展到更复杂的使用场景。

大模型虽然能力非常强,但其弊端就是内容是过时的,这里的过时不是说内容很旧,只是表达内容具有非实时性。比如没有获取天气预报和严重天气警报的能力。因此我们将使用 MCP 来解决这一问题。

构建一个服务器,该服务器提供两个工具:获取警报(get-alerts)和获取预报(get-forecast)。然后,将该服务器连接到 MCP 主机(在本例中为 Claude for Desktop)。

首先我们配置下环境:

(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.
    Each tool specifies its arguments using JSON Schema validation.
    """
    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:
            
            lat_str = 
            lon_str = 
            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
"""Make a request to the NWS API with proper error handling."""
"User-Agent"
"Accept"
"application/geo+json"
try
await
30.0
return
except
return
None
def
format_alert
feature: dict
str
"""Format an alert feature into a concise string."""
"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
""" Handle tool execution requests. Tools can fetch weather data and notify clients of changes. """
if
not
raise
"Missing arguments"
if
"get-alerts"
"state"
if
not
raise
"Missing state parameter"
# Convert state to uppercase to ensure consistent format
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}"
# Format each alert into a concise string
for
in
20
# only take the first 20 alerts
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. Please provide valid numbers for latitude and longitude."
# Basic coordinate validation
if
not
90
90
or
not
180
180
return
type
"text"
"Invalid coordinates. Latitude must be between -90 and 90, longitude between -180 and 180."
async
with
as
# First get the grid point
f"{latitude}"
f"{longitude}"
f"{NWS_API_BASE}/points/{lat_str},{lon_str}"
await
if
not
return
type
"text"
f"Failed to retrieve grid point data for coordinates: {latitude}, {longitude}. This location may not be supported by the NWS API (only US locations are supported)."
# Extract forecast URL from the response
"properties"
"forecast"
if
not
return
type
"text"
"Failed to get forecast URL from grid point data"
# Get the forecast
await
if
not
return
type
"text"
"Failed to retrieve forecast data"
# Format the forecast periods
"properties"
"periods"
if
not
return
type
"text"
"No forecast periods available"
# Format each period into a concise string
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
# Run the server using stdin/stdout streams
async
with
as
await
"weather"
"0.1.0"
if
"__main__"

这段代码中,最核心的其实就是 @server.list_tools() 以及 @server.call_tools() 这两个注解。

@server.list_tools() - 注册用于列出可用工具的处理器 @server.call_tools() - 注册用于执行工具调用的处理器

调用函数的逻辑也比较简单,匹配到对应的工具名称,然后抽取对应的输入参数,然后发起 api 的请求,对获得的结果进行处理:

[图片]

async def main():
    # Run the server using stdin/stdout streams
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream, write_stream, InitializationOptions(
                server_name="weather",
                server_version="0.1.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )

if __name__ == "__main__":
    asyncio.run(main())

(4)服务端与客户端交互

测试服务器与 Claude for Desktop。也给出了构建 MCP 客户端的教程。其中核心的逻辑如下:

async def process_query(self, query: str) -> str:
    """Process a query using Claude and available tools"""
    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]
    # Initial Claude API call
    response = self.anthropic.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1000,
        messages=messages,
        tools=available_tools
    )
    # Process response and handle tool calls
    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
            # Execute tool call
            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 }
                ]
            })
    # Get next response from Claude
    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)

启动客户端,需要打开 Claude for Desktop 应用配置文件:

~/Library/Application Support/Claude/claude_desktop_config.json

如果该文件不存在,确保先创建出来,然后配置以下信息,以示例说明,我们 uv init 的是 weather,所以这里 mcpServers 配置 weather 的服务,args 中的路径设置为你 weather 的绝对路径。

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

保存文件,并重新启动 Claude for Desktop。可以看到 Claude for Desktop 能够识别在天气服务器中暴露的两个工具。

[图片]

[图片]

然后在客户端询问天气,会提示调用 get-forecast 的 tool:

[图片]

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

工具是智能体框架的重要组成部分,允许大模型与外界互动并扩展其能力。即使没有 MCP 协议,也是可以实现 LLM 智能体,只不过存在几个弊端,当有许多不同的 API 时,启用工具使用变得很麻烦,因为任何工具都需要:手动构建 prompt,每当其 API 发生变化时手动更新。

如下图所示:

[图片]

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

整体流程示例:

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

MCP 主机:像 Claude Desktop、IDE 或其他 AI 工具等程序,希望通过 MCP 访问数据。MCP 客户端:与服务器保持 1:1 连接 的协议客户端。

[图片]

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

[图片]

(3)LLM 收到工具处理结果(包括原始的 query 等信息),之后就可以向用户输出最终的答案。

[图片]

总结起来,就一句话,MCP 协议其实是让智能体更容易管理、发现、使用工具。

4. 参考材料

【1】For Server Developers - Model Context Protocol

【2】For Client Developers - Model Context Protocol

【4】MCP 工作原理

目录

  1. 1. 背景
  2. 2. MCP 示例
  3. 3. MCP 到底解决了什么问题
  4. 4. 参考材料

更多推荐文章

查看全部
  • SpringAI Agent 开发实战:基于 Skills 实现代码评审
  • JWT(JSON Web Token)结构化知识体系详解
  • 本地部署大语言模型实践(2):集成外部知识库详细步骤
  • WeTTY 运维监控实战:构建企业级 Web 终端管理
  • GitHub Copilot、Trae 与 Cursor AI 编程工具对比评测
  • FPGA 嵌入式块存储器 RAM:原理与实现指南
  • AI 技术实现电影与小说一键转漫画
  • 前端常用动画库:GSAP、Lottie、Swiper 与 AOS 实战指南
  • C++ 入门指南:历史、首个程序与命名空间详解
  • OpenClaw 对接飞书机器人常见问题与 Gateway 稳定性排查
  • Git 下载、安装与配置教程
  • iOS 运行 Minecraft Java 版指南:PojavLauncher 安装与优化
  • 算法刷题:替换所有问号与提莫攻击(模拟)
  • 具身导航 VLN 最新论文汇总 (2023-2026)
  • 大模型微调与 RAG 的区别是什么?
  • 基于官方 API 搭建 QQ 群聊机器人实战指南
  • Capacitor 实战指南:将 Web 项目打包为跨平台应用
  • OpenVLA 模型微调及机器人端部署实战
  • VSCode GitHub Copilot 安装与使用指南
  • C++备忘录模式:优雅实现对象状态保存与恢复

相关免费在线工具

  • RSA密钥对生成器

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

  • Mermaid 预览与可视化编辑

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

  • 随机西班牙地址生成器

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

  • curl 转代码

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

  • Base64 字符串编码/解码

    将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online

  • Base64 文件转换器

    将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online