Tool Calls

Flows may invoke external tools (e.g., web search, API calls). You can handle them automatically with a handler function or manually by pausing and resuming the iterator.

Automatic mode (`tool_handler`)

Pass a tool_handler function to execute() or steps() to auto-handle all tool calls:

from noukai_sdk import AsyncNoukai
 
def my_tool_handler(tool_calls):
    """
    Called whenever the flow requests tool execution.
    
    Args:
        tool_calls: List of dicts with:
            - id: str (unique ID)
            - function: { name: str, arguments: dict }
    
    Returns:
        List of dicts with:
            - role: "tool"
            - tool_call_id: str
            - content: str (result)
    """
    results = []
    for tc in tool_calls:
        func_name = tc["function"]["name"]
        args = tc["function"]["arguments"]
        
        if func_name == "search":
            content = f"Results for '{args.get('query')}'"
        elif func_name == "get_weather":
            content = f"Weather for {args.get('location')}: 72F"
        else:
            content = f"Unknown tool: {func_name}"
        
        results.append({
            "role": "tool",
            "tool_call_id": tc["id"],
            "content": content
        })
    
    return results
 
async def main():
    async with AsyncNoukai(org="acme", project="research") as client:
        result = await client.flow("web-summarizer").execute(
            message="Summarize TLS 1.3 changes",
            tools=[
                {
                    "type": "function",
                    "function": {
                        "name": "search",
                        "description": "Search the web",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "query": {"type": "string"}
                            }
                        }
                    }
                }
            ],
            tool_handler=my_tool_handler,
        )
        print(result.output)
 
import asyncio
asyncio.run(main())

With tool_handler, the flow executes to completion without pausing. If a tool call fails or returns invalid data, the flow retries automatically (up to 1 retry by default).

Manual mode (`event.resume()`)

Without a tool_handler, the iterator yields ToolCallsRequired when the flow needs tools. Resume manually from inside the iterator loop:

from noukai_sdk import AsyncNoukai, ToolCallsRequired
 
async def main():
    async with AsyncNoukai(org="acme", project="research") as client:
        flow = client.flow("web-summarizer")
        
        async for event in flow.events(
            message="What's new in Python 3.13?",
            tools=[
                {
                    "type": "function",
                    "function": {
                        "name": "search",
                        "description": "Search the web",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "query": {"type": "string", "description": "Search query"}
                            }
                        }
                    }
                }
            ]
            # Note: no tool_handler — we handle manually
        ):
            if isinstance(event, ToolCallsRequired):
                print(f"Flow requested {len(event.tool_calls)} tool calls")
                
                # Execute the tools (your code)
                results = []
                for tc in event.tool_calls:
                    query = tc["function"]["arguments"]["query"]
                    content = my_search(query)  # Your search function
                    results.append({
                        "role": "tool",
                        "tool_call_id": tc["id"],
                        "content": content
                    })
                
                # Resume the flow with results
                await event.resume(tool_results=results)
                print("Flow resumed")
            
            else:
                # Continue processing other events
                if hasattr(event, "output"):
                    print(f"Step: {event.output}")
 
import asyncio
asyncio.run(main())

Manual mode is useful when:

Tool execution is expensive (e.g., real API calls) and you want custom caching/batching
You need to validate or transform tool results before returning them
You're integrating with external orchestration systems

Tool call structure

Tool calls follow OpenAI's format:

{
    "id": "call_1234567890",
    "function": {
        "name": "search",
        "arguments": {
            "query": "climate change"
        }
    }
}

Tool results use the same message format:

{
    "role": "tool",
    "tool_call_id": "call_1234567890",
    "content": "Climate change refers to long-term shifts in global temperatures..."
}

Defining tools

Pass tool definitions to execute() or steps() in OpenAI function-calling format:

tools = [
    {
        "type": "function",
        "function": {
            "name": "search",
            "description": "Search the web for information",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "The search query"
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "Max results to return",
                        "default": 10
                    }
                },
                "required": ["query"]
            }
        }
    }
]
 
result = await flow.execute(
    message="...",
    tools=tools,
    tool_handler=my_handler
)

Error handling

If a tool call fails or returns an error, the flow automatically retries with exponential backoff (1 retry by default). If all retries are exhausted, a ToolCallLimitError is raised.

from noukai_sdk import ToolCallLimitError
 
try:
    result = await flow.execute(message="...", tool_handler=handler)
except ToolCallLimitError as e:
    print(f"Tool calls failed after retries: {e}")

Comparing modes

Aspect	Automatic (`tool_handler`)	Manual (`resume()`)
Simplicity	Very simple — set and forget	Requires loop management
Control	Limited — handler is stateless	Full control over execution
Tool latency	Transparent to your code	You observe and can optimize
Error handling	Automatic retries	You handle errors
Use case	Quick scripts, most flows	Complex logic, real APIs

For most use cases, automatic mode is recommended. Use manual mode when you need advanced control.

Next steps

Iteration patterns: See Step-by-step iteration
Tracing tool calls: See Tracing & debugging
Error handling: See Errors & retries

Automatic mode (tool_handler)

Manual mode (event.resume())