成人免费高清视频,国产精品劲爆视频,色伦专区97中文字幕

# Create a new directory for our project uv init weather cd weather # Create virtual environment and activate it uv venv source .venv/bin/activate # Install dependencies uv add "mcp[cli]" httpx # Create our server file touch server.py

（3）在server.py中構(gòu)建相應(yīng)的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() - 注冊用于執(zhí)行工具調(diào)用的處理器 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())

這段代碼中，最核心的其實(shí)就是@server.list_tools() 以及 @server.call_tool() 這兩個(gè)注解。

調(diào)用函數(shù)的邏輯也比較簡單，匹配到對應(yīng)的工具名稱，然后抽取對應(yīng)的輸入?yún)?shù)，然后發(fā)起api的請求，對獲得的結(jié)果進(jìn)行處理：

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())

測試服務(wù)器與 Claude for Desktop。【2】也給出了構(gòu)建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)

啟動(dòng)客戶端，需要打開 Claude for Desktop 應(yīng)用配置文件：

如果該文件不存在，確保先創(chuàng)建出來，然后配置以下信息，以示例說明，我們uv init的是weather，所以這里mcpServers配置weather的服務(wù)，args中的路徑設(shè)置為你weather的絕對路徑。

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

保存文件，并重新啟動(dòng) Claude for Desktop。可以看到Claude for Desktop 能夠識別在天氣服務(wù)器中暴露的兩個(gè)工具。

然后在客戶端詢問天氣，會(huì)提示調(diào)用get-forecast的tool：

3、MCP到底解決了什么問題

工具是智能體框架的重要組成部分，允許大模型與外界互動(dòng)并擴(kuò)展其能力。即使沒有MCP協(xié)議，也是可以實(shí)現(xiàn)LLM智能體，只不過存在幾個(gè)弊端，當(dāng)有許多不同的 API 時(shí)，啟用工具使用變得很麻煩，因?yàn)槿魏喂ぞ叨夹枰菏謩?dòng)構(gòu)建prompt，每當(dāng)其 API 發(fā)生變化時(shí)手動(dòng)更新【3，4】。

如下圖所示：

MCP其實(shí)解決了當(dāng)存在大量工具時(shí)，能夠自動(dòng)發(fā)現(xiàn)，并自動(dòng)構(gòu)建prompt。

整體流程示例：
（1）以總結(jié)git項(xiàng)目最近5次提交為例，MCP 主機(jī)（與客戶端一起）將首先調(diào)用 MCP 服務(wù)器，詢問有哪些工具可用。

（2）MPC 客戶端接收到所列出的可用工具后，發(fā)給LLM，LLM 收到信息后，可能會(huì)選擇使用某個(gè)工具。它通過主機(jī)向 MCP 服務(wù)器發(fā)送請求，然后接收結(jié)果，包括所使用的工具。

（3）LLM 收到工具處理結(jié)果（包括原始的query等信息），之后就可以向用戶輸出最終的答案。

總結(jié)起來，就一句話，MCP協(xié)議其實(shí)是讓智能體更容易管理、發(fā)現(xiàn)、使用工具。

3、MCP到底解決了什么問題

最新文章