Parallel Tool Calling: Bikin Agent Kamu Berhenti Bolak-Balik Satu Per Satu

Agent kamu lambat karena alasan yang membosankan: dia nanya satu per satu.

Coba tanya "cuaca di Jakarta, Singapore, dan Tokyo gimana?" dan perhatiin apa yang terjadi. Model balas dengan satu tool call. Loop kamu jalanin, append hasilnya, terus tanya lagi. Tiga kota, tiga round trip. Setiap round trip butuh satu network hop plus satu generasi model penuh, jadi jawabannya bisa tiga kali lebih lama dari seharusnya.

API yang kamu pake sekarang udah support beberapa tool call dalam satu respons. OpenAI ngembaliin array tool_calls. Anthropic ngembaliin beberapa block tool_use dalam satu assistant turn. Model sekarang ngelakuin ini by default kalau pertanyaannya cocok. Dokumentasi Anthropic bilang Claude 4 ke atas bikin parallel tool call secara default. Default OpenAI juga ngizinin model manggil beberapa function dalam satu turn, ada switch buat matiin. Kebanyakan loop agent nggak pernah manfaatin ini, karena kebanyakan tutorial nunjukin versi satu-satu.

Perbaikannya kecil: jalanin call-nya secara paralel, dan format hasilnya sesuai yang API harapin. Python copy-paste buat kedua API di bawah, plus kasus di mana jalan berurutan justru pilihan yang bener.

Bentuk respons yang di-batch

OpenAI ngembaliin tool_calls sebagai array di message. Ini contoh dari dokumentasi resmi, dipangkas:

[
  {
    "id": "call_12345xyz",
    "type": "function",
    "function": {
      "name": "get_weather",
      "arguments": "{\"location\": \"Paris, France\"}"
    }
  },
  {
    "id": "call_67890abc",
    "type": "function",
    "function": {
      "name": "get_weather",
      "arguments": "{\"location\": \"Bogotá, Colombia\"}"
    }
  }
]

Dua detail yang penting. arguments itu string JSON, bukan object, jadi parse pake json.loads. Dan tool yang sama bisa muncul dua kali dengan id beda, jadi cocokin hasil berdasarkan id, bukan nama.

Anthropic ngomongin hal yang sama dengan kata-kata yang beda. Responsnya punya stop_reason: "tool_use" dan content berisi beberapa block bertipe tool_use, masing-masing punya id, name, dan input. Kamu balas dengan satu tool_result per block, dicocokin lewat tool_use_id.

Dua-duanya ngasumsi kamu siap nerima beberapa call. Dokumentasi OpenAI bilang gamblang: "it is best practice to assume there are several."

Prerequisites

• Python 3.10+
• pip install openai anthropic
• API key buat provider yang pertama mau kamu test. Kodenya baca OPENAI_API_KEY atau ANTHROPIC_API_KEY dari environment.

Langkah 1: buktiin model kamu bisa batch

Sebelum nyentuh loop, pastiin model beneran ngembaliin beberapa call buat pertanyaan multi-bagian. Script ini ngitung block tool_use dalam satu respons:

from anthropic import Anthropic

client = Anthropic()

tools = [
    {
        "name": "get_weather",
        "description": "Get the current weather in a given location",
        "input_schema": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "City and country, e.g. Jakarta, Indonesia",
                }
            },
            "required": ["location"],
        },
    },
    {
        "name": "get_time",
        "description": "Get the current time in a given timezone",
        "input_schema": {
            "type": "object",
            "properties": {
                "timezone": {
                    "type": "string",
                    "description": "IANA timezone, e.g. Asia/Jakarta",
                }
            },
            "required": ["timezone"],
        },
    },
]

response = client.messages.create(
    model="claude-opus-5",
    max_tokens=1024,
    tools=tools,
    messages=[
        {
            "role": "user",
            "content": "What's the weather in Jakarta and Tokyo, and what time is it in both cities?",
        }
    ],
)

tool_uses = [block for block in response.content if block.type == "tool_use"]
print(f"{len(tool_uses)} tool calls in one response")
for tool in tool_uses:
    print(f"- {tool.name}: {tool.input}")

Kemungkinan besar kamu liat empat call. Kalau cuma satu, model milih jalan berurutan. Bagian troubleshooting di akhir ngebahas kenapa.

Langkah 2: loop Anthropic

Ini pola lengkapnya. Tool di sini read-only, jadi semua call dijalanin barengan pake asyncio.gather:

import asyncio

from anthropic import Anthropic

client = Anthropic()  # reads ANTHROPIC_API_KEY

TOOLS = [
    {
        "name": "get_weather",
        "description": "Get the current weather in a given location",
        "input_schema": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "City and country, e.g. Jakarta, Indonesia",
                }
            },
            "required": ["location"],
        },
    },
    {
        "name": "get_time",
        "description": "Get the current time in a given timezone",
        "input_schema": {
            "type": "object",
            "properties": {
                "timezone": {
                    "type": "string",
                    "description": "IANA timezone, e.g. Asia/Jakarta",
                }
            },
            "required": ["timezone"],
        },
    },
]


def run_tool(name: str, tool_input: dict) -> str:
    """Replace this with a real API call, DB query, or whatever your tool does."""
    if name == "get_weather":
        return f"Sunny, 31C in {tool_input['location']}"
    if name == "get_time":
        return f"14:30 WIB in {tool_input['timezone']}"
    raise ValueError(f"Unknown tool: {name}")


async def execute(tool_use) -> dict:
    await asyncio.sleep(1)  # simulate network latency so tools actually overlap
    try:
        result = run_tool(tool_use.name, tool_use.input)
        return {"type": "tool_result", "tool_use_id": tool_use.id, "content": result}
    except Exception as exc:
        return {
            "type": "tool_result",
            "tool_use_id": tool_use.id,
            "is_error": True,
            "content": str(exc),
        }


async def main():
    messages = [
        {
            "role": "user",
            "content": "What's the weather in Jakarta and Tokyo, and what time is it in both cities?",
        }
    ]

    for _ in range(5):  # hard cap on turns so a buggy loop can't run forever
        response = client.messages.create(
            model="claude-opus-5",
            max_tokens=1024,
            tools=TOOLS,
            messages=messages,
        )

        # The assistant turn is kept verbatim, tool_use blocks included.
        messages.append({"role": "assistant", "content": response.content})

        tool_uses = [block for block in response.content if block.type == "tool_use"]
        if not tool_uses:
            break

        # Run all calls concurrently. Fine for read-only tools.
        results = await asyncio.gather(*[execute(t) for t in tool_uses])

        # All results go in ONE user message, matched by tool_use_id.
        # tool_result blocks must come before any text in this message.
        messages.append({"role": "user", "content": results})

    final_text = next(
        block.text for block in response.content if block.type == "text"
    )
    print(final_text)


if __name__ == "__main__":
    asyncio.run(main())

Aturan formatnya ini yang sering bikin orang salah, jadi hapalin:

• Simpen assistant turn apa adanya. Array content-nya megang block tool_use, dan kamu append apa adanya.
• Balikin semua hasil dalam SATU user message. Satu user message per batch, bukan satu per tool.
• Cocokin tiap hasil pake tool_use_id, dan taruh semua block tool_result sebelum teks apa pun di message itu.
• Kalau kamu skip sebuah call, balikin is_error: true dengan penjelasan singkat, jangan dibuang.

Loop di atas, dijalanin dengan respons mock yang tiap call-nya punya latensi simulasi satu detik, nyelesein empat call dalam sekitar satu detik, bukan empat. Itu intinya.

Langkah 3: loop OpenAI

Idenya sama. tool_calls itu array di message, dan arguments tiap call berupa string JSON yang kamu parse pake json.loads:

import asyncio
import json

from openai import OpenAI

client = OpenAI()  # reads OPENAI_API_KEY

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get the current weather in a given location",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "City and country, e.g. Jakarta, Indonesia",
                    }
                },
                "required": ["location"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_time",
            "description": "Get the current time in a given timezone",
            "parameters": {
                "type": "object",
                "properties": {
                    "timezone": {
                        "type": "string",
                        "description": "IANA timezone, e.g. Asia/Jakarta",
                    }
                },
                "required": ["timezone"],
            },
        },
    },
]


def run_tool(name: str, args: dict) -> str:
    """Replace this with a real API call, DB query, or whatever your tool does."""
    if name == "get_weather":
        return f"Sunny, 31C in {args['location']}"
    if name == "get_time":
        return f"14:30 WIB in {args['timezone']}"
    raise ValueError(f"Unknown tool: {name}")


async def execute(tool_call) -> dict:
    await asyncio.sleep(1)  # simulate network latency so tools actually overlap
    try:
        result = run_tool(
            tool_call.function.name, json.loads(tool_call.function.arguments)
        )
        return {"role": "tool", "tool_call_id": tool_call.id, "content": result}
    except Exception as exc:
        return {
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": f"Error: {exc}",
        }


async def main():
    messages = [
        {"role": "user", "content": "What's the weather in Jakarta and Tokyo?"},
    ]

    for _ in range(5):  # hard cap on turns so a buggy loop can't run forever
        response = client.chat.completions.create(
            model="gpt-5",
            tools=TOOLS,
            messages=messages,
        )
        message = response.choices[0].message

        if not message.tool_calls:
            print(message.content)
            break

        # Keep the assistant turn verbatim, tool_calls included.
        messages.append(message)

        # Run all calls concurrently. Fine for read-only tools.
        results = await asyncio.gather(*[execute(tc) for tc in message.tool_calls])

        # One "tool" message per call, matched by tool_call_id.
        messages.extend(results)


if __name__ == "__main__":
    asyncio.run(main())

Message assistant di-append apa adanya, tool_calls ikut. Terus tiap hasil balik sebagai message tool yang bawa tool_call_id yang cocok. Itu seluruh kontrak OpenAI.

Langkah 4: kapan paralel itu pilihan yang salah

Dokumentasinya tegas. Operasi read-only yang independen aman dijalanin paralel buat ngurangin latency. Tool yang punya side effect, shared state, atau butuh urutan tertentu, lebih baik dijalanin berurutan.

Side effect itu kasus klasiknya: send_email, write_file, debit_account. Jalanin barengan risikonya dobel kirim dan race condition. Kalau model tetep batch dua-duanya, jalanin berurutan dan berhenti di kegagalan pertama. Balikin is_error: true buat call yang kamu skip, dengan catatan singkat kayak "Not executed: the preceding write_file call failed." Model bakal minta ulang di turn berikutnya.

Call yang saling tergantung juga gitu. Tool B butuh output tool A, dan dua-duanya dateng dalam satu batch. Jalanin urut, berhenti kalau ada yang gagal. Biar batch yang tergantung jarang muncul, tambahin ke system prompt: "Only batch tool calls that are independent of each other."

Langkah 5: kontrol paralelisme

OpenAI:

Parameter	Efek
parallel_tool_calls: false	nol atau satu tool per turn
tool_choice: "required"	satu atau lebih tool
tool_choice: {"type": "function", "name": "get_weather"}	persis tool itu
tool_choice: "none"	nggak ada tool sama sekali

Anthropic:

Parameter	Efek
tool_choice: {"type": "auto", "disable_parallel_tool_use": true}	maksimal satu tool per turn, jawaban teks biasa tetep boleh
{"type": "any"} atau {"type": "tool", ...} plus disable_parallel_tool_use: true	persis satu tool

Satu jebakan: disable_parallel_tool_use ditaruh di dalem objek tool_choice, bukan di level atas request.

Kapan matiin? API yang kena rate limit. Tool yang mutasi shared state. Atau pas kamu butuh trace langkah-demi-langkah buat audit dan tiap call harus bisa diurutkan.

Jebakan yang bakal ngegigit

Format hasil yang salah bikin paralelisme mati. Ini penyebab nomor satu. Kirim tiap tool result sebagai user message terpisah, model bakal belajar jalan berurutan. Semua hasil masuk dalam SATU user message, hasil duluan sebelum teks. OpenAI idenya sama: satu message tool per tool_call_id, dan balikin semua id, termasuk yang gagal. Taruh teks error-nya di content.

Rate limit. asyncio.gather ngegas semua barengan. Sepuluh tool artinya sepuluh HTTP call konkuren. Bungkus eksekusinya pake asyncio.Semaphore(3) kalau tool kamu nyentuh API pihak ketiga.

Skema tool makan token tiap request. OpenAI nyuntikin definisi function ke system message dan nge-bill sebagai input token. Jaga deskripsi tetap pendek, dan jangan bawa dua puluh tool kalau lima aja cukup. Kalau perlu ngebatesin tool mana yang bisa dipanggil tanpa ngedrop skemanya, allowed_tools bisa, dan ini juga bikin prompt caching tetep jalan.

Output tool itu nggak bisa dipercaya. Hasilnya bisa bawa instruksi yang disuntikin, polanya disebut indirect prompt injection. Simpen di dalem block tool_result dan jangan pernah echo ke system prompt.

Streaming ngubah bentuk. tool_calls dateng sebagai delta yang dikunci per index. Gabungin per index sampe stream selesai, baru jalanin loop gather yang sama.

Ukur

Catet rata-rata jumlah tool call per assistant turn. Di atas 1.0 artinya batching jalan. Kalau mentok di 1.0, berarti format atau prompting yang bermasalah:

tool_call_messages = [
    msg for msg in messages
    if any(block.type == "tool_use" for block in msg.content)
]
total_calls = sum(
    len([b for b in msg.content if b.type == "tool_use"])
    for msg in tool_call_messages
)
print(f"avg tools per message: {total_calls / len(tool_call_messages):.2f}")

Langkah selanjutnya

• strict: true di OpenAI maksa tiap call cocok sama skemanya.
• Tool Runner di SDK Anthropic ngurusin seluruh loop ini buat kamu, termasuk error wrapping, kalau nggak mau maintain sendiri.
• Kalau kamu bangun di atas MCP, loop yang sama berlaku. Host nerima beberapa tool call dan ngejalaninnya dengan cara yang sama.

Saran saya: mulai dari loop manual biar kamu beneran paham aturan formatnya, ukur, baru putusin apakah butuh abstraksi.

Intinya kecil: anggap aja tiap turn bisa bawa beberapa call, jalanin yang independen secara paralel, dan jaga format hasil sesuai yang API harapin. Itu aja udah ngilangin sebagian besar latency di loop agent yang naif.

Referensi

• OpenAI: Function calling guide
• Anthropic: Parallel tool use
• Anthropic: Handle tool calls