Parallel tool execution

When an agent needs weather, stock price, and calendar data, calling them sequentially wastes time. This recipe shows how to request multiple tool calls in parallel from a single model response, execute them concurrently, and return all results in one follow-up message.

The script

parallel_tools.py

import os
import json
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["RADIUM_API_KEY"],
    base_url="https://api.radium.cloud/v1",
)


def get_weather(location: str) -> str:
    return f"Sunny, 22°C in {location}"


def get_stock(ticker: str) -> str:
    prices = {"AAPL": 185.5, "TSLA": 240.0, "GOOGL": 140.0}
    return f"{ticker} is trading at ${prices.get(ticker, 'N/A')}"


def get_calendar(date: str) -> str:
    return f"2 meetings on {date}: 10:00 AM standup, 2:00 PM review"


TOOL_REGISTRY = {
    "get_weather": get_weather,
    "get_stock": get_stock,
    "get_calendar": get_calendar,
}

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get the current weather for a location.",
            "parameters": {
                "type": "object",
                "properties": {"location": {"type": "string"}},
                "required": ["location"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_stock",
            "description": "Get the current stock price for a ticker symbol.",
            "parameters": {
                "type": "object",
                "properties": {"ticker": {"type": "string"}},
                "required": ["ticker"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_calendar",
            "description": "Get calendar events for a date.",
            "parameters": {
                "type": "object",
                "properties": {"date": {"type": "string"}},
                "required": ["date"],
            },
        },
    },
]


def run_agent(user_message: str) -> str:
    messages = [
        {"role": "system", "content": "You are a helpful assistant with access to external tools."},
        {"role": "user", "content": user_message},
    ]

    response = client.chat.completions.create(
        model="clarke-1.0",
        messages=messages,
        tools=TOOLS,
        tool_choice="auto",
        max_tokens=1024,
    )

    message = response.choices[0].message
    if not message.tool_calls:
        return message.content

    tool_results = []
    for call in message.tool_calls:
        fn = TOOL_REGISTRY[call.function.name]
        args = json.loads(call.function.arguments)
        result = fn(**args)
        tool_results.append({
            "tool_call_id": call.id,
            "role": "tool",
            "content": result,
        })

    messages.append(message.model_dump())
    for tr in tool_results:
        messages.append({
            "role": "tool",
            "tool_call_id": tr["tool_call_id"],
            "content": tr["content"],
        })

    final = client.chat.completions.create(
        model="clarke-1.0",
        messages=messages,
        max_tokens=1024,
    )
    return final.choices[0].message.content


# --- Run it ---
query = "What's the weather in San Francisco, the AAPL stock price, and what's on my calendar for 2024-12-25?"
print(f"User: {query}\n")
print(f"Agent: {run_agent(query)}")

Run it

export RADIUM_API_KEY="YOUR_RADIUM_API_KEY"
python parallel_tools.py

Sample output

User: What's the weather in San Francisco, the AAPL stock price, and what's on my calendar for 2024-12-25?

Agent: The weather in San Francisco is sunny and 22°C. AAPL is trading at $185.5. On 2024-12-25 you have 2 meetings: 10:00 AM standup and 2:00 PM review.

Async version for slow tools

If tools hit external APIs with latency, execute them concurrently with asyncio.gather:

async def run_agent_async(user_message: str) -> str:
    response = client.chat.completions.create(
        model="clarke-1.0", messages=[...], tools=TOOLS, tool_choice="auto"
    )
    message = response.choices[0].message
    if not message.tool_calls:
        return message.content

    async def execute_call(call):
        fn = TOOL_REGISTRY[call.function.name]
        args = json.loads(call.function.arguments)
        result = await fn(**args) if asyncio.iscoroutinefunction(fn) else fn(**args)
        return {"tool_call_id": call.id, "content": result}

    results = await asyncio.gather(*[execute_call(c) for c in message.tool_calls])
    # append results to messages and get final response

Tips

clarke-1.0 handles parallel tool intent better than smaller models.
Structure tool names clearly — models rely on name and description to pick the right tool.
Keep descriptions short but include example inputs so arguments are formatted correctly.
Limit to 5-8 tools per call — more causes the model to miss or confuse tools.
Log which tools were called per request — it’s the #1 debugging signal for agent behavior.

Getting started

Recipes

Parallel tool execution

Parallel tool execution

The script

Run it

Sample output

Async version for slow tools

Tips

Next steps

Tool calling agent

Structured extraction

​Parallel tool execution

​The script

​Run it

​Sample output

​Async version for slow tools

​Tips

​Next steps

Tool calling agent

Structured extraction

Parallel tool execution

The script

Run it

Sample output

Async version for slow tools

Tips

Next steps