【大模型实战篇】基于Claude MCP协议的智能体落地示例

【大模型实战篇】基于Claude MCP协议的智能体落地示例

1. 背景

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

        今天我们基于claude的官方例子--获取天气预报【1】,来理解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.list_tools() - 注册用于列出可用工具的处理器 #@server.call_tool() - 注册用于执行工具调用的处理器 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={ "type": "object", "properties": { "state": { "type": "string", "description": "Two-letter state code (e.g. CA, NY)", }, }, "required": ["state"], }, ), types.Tool( name="get-forecast", description="Get weather forecast for a location", inputSchema={ "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.""" headers = { "User-Agent": USER_AGENT, "Accept": "application/geo+json" } try: response = await client.get(url, headers=headers, timeout=30.0) response.raise_for_status() return response.json() except Exception: return None def format_alert(feature: dict) -> str: """Format an alert feature into a concise string.""" props = feature["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[types.TextContent | types.ImageContent | types.EmbeddedResource]: """ Handle tool execution requests. Tools can fetch weather data and notify clients of changes. """ if not arguments: raise ValueError("Missing arguments") if name == "get-alerts": state = arguments.get("state") if not state: raise ValueError("Missing state parameter") # Convert state to uppercase to ensure consistent format state = state.upper() if len(state) != 2: raise ValueError("State must be a two-letter code (e.g. CA, NY)") async with httpx.AsyncClient() as client: alerts_url = f"{NWS_API_BASE}/alerts?area={state}" alerts_data = await make_nws_request(client, alerts_url) if not alerts_data: return [types.TextContent(type="text", text="Failed to retrieve alerts data")] features = alerts_data.get("features", []) if not features: return [types.TextContent(type="text", text=f"No active alerts for {state}")] # Format each alert into a concise string formatted_alerts = [format_alert(feature) for feature in features[:20]] # only take the first 20 alerts alerts_text = f"Active alerts for {state}:\n\n" + "\n".join(formatted_alerts) return [ types.TextContent( type="text", text=alerts_text ) ] elif name == "get-forecast": try: latitude = float(arguments.get("latitude")) longitude = float(arguments.get("longitude")) except (TypeError, ValueError): return [types.TextContent( type="text", text="Invalid coordinates. Please provide valid numbers for latitude and longitude." )] # Basic coordinate validation if not (-90 <= latitude <= 90) or not (-180 <= longitude <= 180): return [types.TextContent( type="text", text="Invalid coordinates. Latitude must be between -90 and 90, longitude between -180 and 180." )] async with httpx.AsyncClient() as client: # First get the grid point lat_str = f"{latitude}" lon_str = f"{longitude}" points_url = f"{NWS_API_BASE}/points/{lat_str},{lon_str}" points_data = await make_nws_request(client, points_url) if not points_data: return [types.TextContent(type="text", 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 = points_data.get("properties", {}) forecast_url = properties.get("forecast") if not forecast_url: return [types.TextContent(type="text", text="Failed to get forecast URL from grid point data")] # Get the forecast forecast_data = await make_nws_request(client, forecast_url) if not forecast_data: return [types.TextContent(type="text", text="Failed to retrieve forecast data")] # Format the forecast periods periods = forecast_data.get("properties", {}).get("periods", []) if not periods: return [types.TextContent(type="text", text="No forecast periods available")] # Format each period into a concise string formatted_forecast = [] for period in periods: forecast_text = ( 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" "---" ) formatted_forecast.append(forecast_text) forecast_text = f"Forecast for {latitude}, {longitude}:\n\n" + "\n".join(formatted_forecast) return [types.TextContent( type="text", text=forecast_text )] else: raise ValueError(f"Unknown tool: {name}") 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={}, ), ), ) # This is needed if you'd like to connect to a custom client if __name__ == "__main__": asyncio.run(main()) 

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

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

        调用函数的逻辑也比较简单,匹配到对应的工具名称,然后抽取对应的输入参数,然后发起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={}, ), ), ) # This is needed if you'd like to connect to a custom client if __name__ == "__main__": asyncio.run(main())

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

        测试服务器与 Claude for Desktop。【2】也给出了构建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 发生变化时手动更新【3,4】。

        如下图所示:

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

        整体流程示例:

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

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

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

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

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

4. 参考材料

【1】For Server Developers - Model Context Protocol

【2】For Client Developers - Model Context Protocol

【3】AI Agent框架综述

【4】MCP工作原理

Read more

微博爬虫Web管理界面完全配置手册

还在为复杂的爬虫配置而烦恼吗?weibospider项目的Web管理界面让你摆脱繁琐的命令行操作,通过直观的图形界面轻松管理整个微博数据采集系统。本文将带你从零开始,全面掌握这个基于Django Admin的强大管理平台。 【免费下载链接】weibospider:zap: A distributed crawler for weibo, building with celery and requests. 项目地址: https://gitcode.com/gh_mirrors/wei/weibospider 🎯 为什么选择Web管理界面 传统的爬虫配置往往需要通过修改配置文件、执行复杂命令来完成,这不仅效率低下,还容易出错。weibospider的Web管理界面将这些复杂操作封装成简单的表单和按钮,让配置工作变得轻松愉快。 核心优势: * 🖥️ 可视化操作,告别命令行 * ⚡ 批量处理,效率翻倍 * 📊 实时监控,掌握运行状态 * 🔧 灵活配置,满足个性化需求 🛠️ 环境搭建与初始化 数据库配置调整 首先打开admin/weibo_admin/s

By Ne0inhk

告别复杂配置:AWPortrait-Z云端WebUI一站式解决方案

告别复杂配置:AWPortrait-Z云端WebUI一站式解决方案 你是不是也遇到过这种情况?看到同行的电商商品图里,模特皮肤细腻、光影自然、背景干净,点击率和转化率蹭蹭上涨。再看看自己的产品图,总觉得差了点“高级感”。你想用AI来优化人像图,听说AWPortrait-Z效果特别好——能自动修复皮肤噪点、重塑光影、提升画质细节,甚至还能智能换脸生成专属模特形象。 可当你兴致勃勃地准备动手时,却发现:安装依赖报错、CUDA版本不匹配、PyTorch编译失败、模型路径找不到……折腾一整天,连界面都没跑起来。这正是电商店主老王的真实经历。他试了三天,装了卸、卸了装,最后只能放弃:“搞技术太难了,我们小商家哪有时间研究这些?” 别担心,今天我要告诉你一个好消息:现在你完全不需要自己配置环境!通过ZEEKLOG星图平台提供的 AWPortrait-Z云端WebUI镜像,只需点击几下,就能在云端直接使用这个强大的AI人像增强工具。无需懂代码、不用装驱动、不怕版本冲突,真正实现“开箱即用”。 这篇文章就是为像老王这样的小白用户量身打造的。我会手把手带你完成从零到出图的全过程,让你5分钟内就能

By Ne0inhk
快过年了,写个游戏玩玩,放松下,解析俄罗斯方块游戏(可直接复制代码使用,玩游戏)。罗斯方块游戏技术解析:从前端实现到工程化思考

快过年了,写个游戏玩玩,放松下,解析俄罗斯方块游戏(可直接复制代码使用,玩游戏)。罗斯方块游戏技术解析:从前端实现到工程化思考

前言:哈喽,大家好,今天给大家分享一篇文章!并提供具体代码帮助大家深入理解,彻底掌握!创作不易,如果能帮助到大家或者给大家一些灵感和启发,欢迎点赞 + 收藏 + 关注哦 💕 快过年了,写个游戏玩玩,放松下,解析俄罗斯方块游戏(可直接复制代码,玩游戏)。罗斯方块游戏技术解析:从前端实现到工程化思考 📚 本文简介 本文解析了一个基于HTML5+CSS3+JavaScript的俄罗斯方块网页游戏实现。项目采用模块化设计,包含index.html、style.css和script.js三个核心文件,遵循前端开发最佳实践。HTML结构采用语义化布局,使用Canvas双画布分别渲染主游戏区和预览区。CSS运用Flexbox布局、毛玻璃效果、过渡动画等现代特性,实现响应式设计。JavaScript处理游戏逻辑,包括方块旋转、碰撞检测等核心算法。项目兼顾性能与用户体验,是前端游戏开发的经典案例。全文从架构设计到实现细节进行了深度技术解析。 目录 * 快过年了,写个游戏玩玩,放松下,解析俄罗斯方块游戏(可直接复制代码,玩游戏)。罗斯方块游戏技术解析:

By Ne0inhk
速通前端篇 —— CSS

速通前端篇 —— CSS

找往期文章包括但不限于本期文章中不懂的知识点: 个人主页:我要学编程程(ಥ_ಥ)-ZEEKLOG博客 所属专栏:速通前端 目录 CSS的介绍 基本语法规范 CSS选择器 标签选择器 class选择器  id选择器  复合选择器  通配符选择器 CSS常见样式  颜色 color 字体大小 font-size  边框 border  宽度 与 高度  内边距 外边距  CSS的介绍 CSS(Cascading Style Sheet),层叠样式表,用于控制页面的样式。CSS能够对网页中元素位置的排版进行像素级精确控制,实现美化页面的效果。能够做到页面的样式和结构分离。简单理解,CSS就是类似于对页面进行"化妆",让页面变得更加好看。 基本语法规范 选择器+{一条/N条声明} 1、

By Ne0inhk