Parallel tool calling:
несколько вызовов за раз

import json
import anthropic

client = anthropic.Anthropic()

def get_weather(city: str) -> dict:
    # Заглушка — в реальности HTTP-запрос к weather API
    data = {"Москва": (-3, "Снег"), "Лондон": (8, "Дождь"), "Токио": (12, "Ясно")}
    temp, desc = data.get(city, (0, "Неизвестно"))
    return {"city": city, "temp": temp, "description": desc}

TOOLS = [{
    "name": "get_weather",
    "description": "Возвращает текущую погоду в указанном городе.",
    "input_schema": {
        "type": "object",
        "properties": {
            "city": {"type": "string", "description": "Название города на русском языке"}
        },
        "required": ["city"]
    }
}]

HANDLERS = {"get_weather": get_weather}

def run_agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    for _ in range(10):
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=1024,
            tools=TOOLS,
            messages=messages
        )

        if response.stop_reason == "end_turn":
            return response.content[0].text

        if response.stop_reason == "tool_use":
            # Собираем результаты ВСЕХ tool_use блоков
            tool_results = []
            for block in response.content:
                if block.type != "tool_use":
                    continue
                handler = HANDLERS.get(block.name)
                try:
                    result = handler(**block.input) if handler else {"error": f"unknown tool: {block.name}"}
                except Exception as e:
                    result = {"error": str(e)}

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,         # каждый ID уникален
                    "content": json.dumps(result, ensure_ascii=False)
                })

            # Возвращаем ВСЕ результаты в одном сообщении
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user",      "content": tool_results})

    raise RuntimeError("Превышен лимит итераций")

print(run_agent("Сравни погоду в Москве, Лондоне и Токио"))
# Модель сгенерирует 3 tool_use блока за один ответ

import json
import asyncio
import anthropic

client = anthropic.AsyncAnthropic()    # async-клиент

# Async-версии инструментов
async def get_weather(city: str) -> dict:
    await asyncio.sleep(0)             # в реальности: await httpx.AsyncClient().get(...)
    data = {"Москва": (-3, "Снег"), "Лондон": (8, "Дождь"), "Токио": (12, "Ясно")}
    temp, desc = data.get(city, (0, "Неизвестно"))
    return {"city": city, "temp": temp, "description": desc}

ASYNC_HANDLERS = {"get_weather": get_weather}

async def execute_tools_concurrently(tool_use_blocks: list) -> list:
    """Запускаем все функции конкурентно через asyncio.gather"""
    async def run_one(block):
        handler = ASYNC_HANDLERS.get(block.name)
        try:
            result = await handler(**block.input) if handler else {"error": f"unknown: {block.name}"}
        except Exception as e:
            result = {"error": str(e)}
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": json.dumps(result, ensure_ascii=False)
        }

    # Все функции стартуют одновременно — ждём самую медленную
    return await asyncio.gather(*[run_one(b) for b in tool_use_blocks])


async def run_agent_async(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    for _ in range(10):
        response = await client.messages.create(
            model="claude-opus-4-6",
            max_tokens=1024,
            tools=[{
                "name": "get_weather",
                "description": "Погода в городе.",
                "input_schema": {
                    "type": "object",
                    "properties": {"city": {"type": "string", "description": "Город на русском"}},
                    "required": ["city"]
                }
            }],
            messages=messages
        )

        if response.stop_reason == "end_turn":
            return response.content[0].text

        if response.stop_reason == "tool_use":
            tool_blocks = [b for b in response.content if b.type == "tool_use"]

            # Конкурентное выполнение: все функции параллельно
            tool_results = await execute_tools_concurrently(tool_blocks)

            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user",      "content": tool_results})

    raise RuntimeError("Превышен лимит итераций")


# Запуск
result = asyncio.run(run_agent_async("Сравни погоду в Москве, Лондоне и Токио"))
print(result)