from typing import Anyimport httpxfrom claude_agent_sdk import tool, create_sdk_mcp_server# Define a tool: name, description, input schema, handler@tool( "get_temperature", "Get the current temperature at a location", {"latitude": float, "longitude": float},)async def get_temperature(args: dict[str, Any]) -> dict[str, Any]: async with httpx.AsyncClient() as client: response = await client.get( "https://api.open-meteo.com/v1/forecast", params={ "latitude": args["latitude"], "longitude": args["longitude"], "current": "temperature_2m", "temperature_unit": "fahrenheit", }, ) data = response.json() # Return a content array - Claude sees this as the tool result return { "content": [ { "type": "text", "text": f"Temperature: {data['current']['temperature_2m']}°F", } ] }# Wrap the tool in an in-process MCP serverweather_server = create_sdk_mcp_server( name="weather", version="1.0.0", tools=[get_temperature],)
import asynciofrom claude_agent_sdk import query, ClaudeAgentOptions, ResultMessageasync def main(): options = ClaudeAgentOptions( mcp_servers={"weather": weather_server}, allowed_tools=["mcp__weather__get_temperature"], ) async for message in query( prompt="What's the temperature in San Francisco?", options=options, ): # ResultMessage is the final message after all tool calls complete if isinstance(message, ResultMessage) and message.subtype == "success": print(message.result)asyncio.run(main())
# Define a second tool for the same server@tool( "get_precipitation_chance", "Get the hourly precipitation probability for a location. " "Optionally pass 'hours' (1-24) to control how many hours to return.", {"latitude": float, "longitude": float},)async def get_precipitation_chance(args: dict[str, Any]) -> dict[str, Any]: # 'hours' isn't in the schema - read it with .get() to make it optional hours = args.get("hours", 12) async with httpx.AsyncClient() as client: response = await client.get( "https://api.open-meteo.com/v1/forecast", params={ "latitude": args["latitude"], "longitude": args["longitude"], "hourly": "precipitation_probability", "forecast_days": 1, }, ) data = response.json() chances = data["hourly"]["precipitation_probability"][:hours] return { "content": [ { "type": "text", "text": f"Next {hours} hours: {'%, '.join(map(str, chances))}%", } ] }# Rebuild the server with both tools in the arrayweather_server = create_sdk_mcp_server( name="weather", version="1.0.0", tools=[get_temperature, get_precipitation_chance],)
from claude_agent_sdk import tool, ToolAnnotations@tool( "get_temperature", "Get the current temperature at a location", {"latitude": float, "longitude": float}, annotations=ToolAnnotations( readOnlyHint=True ), # Lets Claude batch this with other read-only calls)async def get_temperature(args): return {"content": [{"type": "text", "text": "..."}]}
import jsonimport httpxfrom typing import Any@tool( "fetch_data", "Fetch data from an API", {"endpoint": str}, # Simple schema)async def fetch_data(args: dict[str, Any]) -> dict[str, Any]: try: async with httpx.AsyncClient() as client: response = await client.get(args["endpoint"]) if response.status_code != 200: # Return the failure as a tool result so Claude can react to it. # is_error marks this as a failed call rather than odd-looking data. return { "content": [ { "type": "text", "text": f"API error: {response.status_code} {response.reason_phrase}", } ], "is_error": True, } data = response.json() return {"content": [{"type": "text", "text": json.dumps(data, indent=2)}]} except Exception as e: # Composes the message Claude reads. An uncaught exception would # reach Claude as the raw str(e) with no context. return { "content": [{"type": "text", "text": f"Failed to fetch data: {str(e)}"}], "is_error": True, }
import base64import httpx# Define a tool that fetches an image from a URL and returns it to Claude@tool("fetch_image", "Fetch an image from a URL and return it to Claude", {"url": str})async def fetch_image(args): async with httpx.AsyncClient() as client: # Fetch the image bytes response = await client.get(args["url"]) return { "content": [ { "type": "image", "data": base64.b64encode(response.content).decode( "ascii" ), # Base64-encode the raw bytes "mimeType": response.headers.get( "content-type", "image/png" ), # Read MIME type from the response } ] }
资源
资源块嵌入由 URI 标识的内容片段。URI 是 Claude 引用的标签；实际内容位于块的 text 或 blob 字段中。当您的工具生成稍后按名称寻址有意义的内容时使用此功能，例如生成的文件或来自外部系统的记录。
此示例显示从工具处理程序内部返回的资源块。URI file:///tmp/report.md 是 Claude 可以稍后引用的标签；SDK 不从该路径读取。
Copy
return { content: [ { type: "resource", resource: { uri: "file:///tmp/report.md", // Label for Claude to reference, not a path the SDK reads mimeType: "text/markdown", text: "# Report\n..." // The actual content, inline } } ]};
桂公网安备45010502001169号