from typing import Any
import httpx
from claude_agent_sdk import tool, create_sdk_mcp_server


# 定义工具：名称、描述、输入 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 {
        "content": [
            {
                "type": "text",
                "text": f"Temperature: {data['current']['temperature_2m']}°F",
            }
        ]
    }


# 将工具包装在进程内 MCP 服务器中
weather_server = create_sdk_mcp_server(
    name="weather",
    version="1.0.0",
    tools=[get_temperature],
)

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage


async def main():
    options = ClaudeAgentOptions(
        mcp_servers={"weather": weather_server},
        allowed_tools=["mcp__weather__get_temperature"],
    )

    async for message in query(
        prompt="旧金山的温度是多少？",
        options=options,
    ):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)


asyncio.run(main())

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  # 允许 Claude 将此工具与其他只读调用批量执行
    ),
)
async def get_temperature(args):
    return {"content": [{"type": "text", "text": "..."}]}

import httpx
from typing import Any


@tool(
    "fetch_data",
    "Fetch data from an API",
    {"endpoint": str},
)
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:
                # 返回错误结果让 Claude 能做出反应，而非直接抛出异常
                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": str(data)}]}
    except Exception as e:
        # 捕获异常保持循环存活，未捕获异常会终止整个 query()
        return {
            "content": [{"type": "text", "text": f"Failed to fetch data: {str(e)}"}],
            "is_error": True,
        }

import base64
import httpx


@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:
        response = await client.get(args["url"])

    return {
        "content": [
            {
                "type": "image",
                "data": base64.b64encode(response.content).decode("ascii"),
                "mimeType": response.headers.get("content-type", "image/png"),
            }
        ]
    }

options = ClaudeAgentOptions(
    mcp_servers={
        "github": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-github"],
            "env": {"GITHUB_TOKEN": os.environ["GITHUB_TOKEN"]},
        }
    },
    allowed_tools=["mcp__github__list_issues", "mcp__github__search_issues"],
)

options = ClaudeAgentOptions(
    mcp_servers={
        "remote-api": {
            "type": "sse",
            "url": "https://api.example.com/mcp/sse",
            "headers": {"Authorization": f"Bearer {os.environ['API_TOKEN']}"},
        }
    },
    allowed_tools=["mcp__remote-api__*"],
)

import asyncio
import os
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage


async def main():
    connection_string = os.environ["DATABASE_URL"]

    options = ClaudeAgentOptions(
        mcp_servers={
            "postgres": {
                "command": "npx",
                "args": [
                    "-y",
                    "@modelcontextprotocol/server-postgres",
                    connection_string,
                ],
            }
        },
        # 只允许读查询，不允许写
        allowed_tools=["mcp__postgres__query"],
    )

    # 自然语言查询 — Claude 自动编写 SQL
    async for message in query(
        prompt="上周有多少用户注册？按天分组。",
        options=options,
    ):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)


asyncio.run(main())

from claude_agent_sdk import query, ClaudeAgentOptions, SystemMessage, ResultMessage


async for message in query(prompt="处理数据", options=options):
    if isinstance(message, SystemMessage) and message.subtype == "init":
        failed_servers = [
            s
            for s in message.data.get("mcp_servers", [])
            if s.get("status") != "connected"
        ]

        if failed_servers:
            print(f"连接失败: {failed_servers}")

    if isinstance(message, ResultMessage) and message.subtype == "error_during_execution":
        print("执行失败")

from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

options = ClaudeAgentOptions(
    mcp_servers={
        "enterprise-tools": {
            "type": "http",
            "url": "https://tools.example.com/mcp",
        }
    },
    allowed_tools=["mcp__enterprise-tools__*"],  # 通配符预批准该服务器所有工具
    env={
        "ENABLE_TOOL_SEARCH": "auto:5"  # 工具定义超过上下文 5% 时激活
    },
)

async for message in query(
    prompt="查找并运行适当的数据库查询",
    options=options,
):
    if isinstance(message, ResultMessage) and message.subtype == "success":
        print(message.result)