Agent SDK · Tools and sessions

Client tools.

Local tool handlers the agent can invoke mid-turn, wired into the roundtrip loop.

Client tools are handlers you register that the agent can invoke mid-turn. The SDK handles the entire roundtrip loop: detecting tool calls, executing your handler, feeding the result back to the agent, and repeating until the agent stops calling tools or the roundtrip limit is reached.

Tool execution loop

text

┌─────────────────────────────────────────────────────────┐
│  Turn 0                                                 │
│  1. Send messages + client_tools to server              │
│  2. Stream events — agent responds, may call tools      │
│  3. If tool calls detected:                             │
│     a. client.tool.requested → client.tool.executing    │
│     b. Execute handler from tool_handlers               │
│     c. client.tool.result                               │
│     d. Append result to messages as a user message      │
│     e. client.tool.roundtrip (turn + 1 / max_turns)     │
│     f. Go to Turn 1                                     │
│  4. If no tool calls: run is done                       │
│  5. If turn >= max_tool_roundtrips: truncated           │
└─────────────────────────────────────────────────────────┘

Declaring and handling tools

Python

import asyncio
import json

from matilda_agent_sdk import ToolResult, stream

# 1. Declare the tools so the server knows they exist
client_tools = [
    {"name": "get_weather", "description": "Get current weather for a city", "parameters": {"type": "object"}},
    {"name": "calculate", "description": "Evaluate a math expression", "parameters": {"type": "object"}},
]


# 2. Register handlers — the SDK calls these when the agent invokes a tool
async def get_weather(args, ctx):
    city = args.get("city", "unknown")
    return ToolResult(content=json.dumps({"city": city, "temp": 22, "condition": "sunny"}))


async def calculate(args, ctx):
    try:
        # NOTE: eval is a stand-in for a real expression parser — never eval
        # untrusted input in production.
        result = eval(str(args["expression"]), {"__builtins__": {}}, {})
        return ToolResult(content=str(result))
    except Exception:
        return ToolResult(content="Invalid expression", is_error=True)


tool_handlers = {"get_weather": get_weather, "calculate": calculate}


# 3. Pass both to stream() or run()
async def main():
    async for event in stream(
        {"name": "assistant", "instructions": "Use the available tools to answer questions."},
        "What is the weather in Sydney, and what is 15 * 23?",
        tool_handlers=tool_handlers,
        client_tools=client_tools,
    ):
        if event.type == "client.tool.requested":
            print(f"→ Agent requested: {event.name}({json.dumps(event.args)})")
        if event.type == "client.tool.executing":
            print(f"⚙ Executing: {event.name}")
        if event.type == "client.tool.result":
            suffix = " (error)" if event.is_error else ""
            print(f"← Result: {event.result}{suffix}")
        if event.type == "client.tool.roundtrip":
            print(f"  Roundtrip {event.turn}/{event.max_turns}")
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)


asyncio.run(main())

`ToolHandler`

Python

ToolHandler = Callable[[dict, ToolExecutionContext], Awaitable[ToolResult]]

`ToolExecutionContext`

Python

@dataclass(frozen=True)
class ToolExecutionContext:
    tool_call_id: str | None = None

`ToolResult`

Python

@dataclass(frozen=True)
class ToolResult:
    content: str
    is_error: bool = False

`ToolHandlers`

Python

ToolHandlers = dict[str, ToolHandler]

`max_tool_roundtrips`

Controls how many back-and-forth tool cycles the SDK allows before stopping. Default is 25 (DEFAULT_MAX_TOOL_ROUNDTRIPS). Lower it to prevent infinite loops or control cost.

Python

result = await runner.run(
    {"name": "tool-heavy-agent", "instructions": "Use tools to gather information."},
    "Do a task that needs tools",
    max_tool_roundtrips=5,
    tool_handlers={"search": lambda args, ctx: search_handler(args)},
    client_tools=[{"name": "search", "description": "Search", "parameters": {"type": "object"}}],
)

roundtrips = [e for e in result.events if e.type == "client.tool.roundtrip"]
print(f"Roundtrips used: {len(roundtrips)} (max was 5)")

Advertised-tool guard

The SDK enforces that the agent can only call tools that were advertised for the current turn. If the model calls a tool that wasn't in client_tools, the SDK returns an error result instead of executing a handler:

text

Tool not offered this turn: <name>

This prevents a model from talking the runner into invoking a handler that was never offered — tool output is untrusted input.

DSML tool-call interception

Some models emit tool calls as text tokens wrapped in DSML markup (<｜DSML｜tool_call>{...}</｜DSML｜tool_call>) instead of using native function calling. The SDK automatically intercepts these text tokens, parses the JSON payload, and surfaces them as native ClientToolRequested events — the consumer never sees the raw markup.

This interception is fully automatic and applies to all streaming paths (the tag constants are exported as DSML_TOOL_CALL_OPEN / DSML_TOOL_CALL_CLOSE).

Human-in-the-loop

The tool execution loop makes human-in-the-loop trivial: a tool handler is just an async callable, so it can block on stdin, a UI prompt, or any other input source.

Python

import asyncio

from matilda_agent_sdk import ToolResult


async def ask_user(args, ctx):
    answer = await asyncio.to_thread(input, f"\n  Agent asks: {args['question']}\n  > ")
    return ToolResult(content=answer.strip())


tool_handlers = {"ask_user": ask_user}
client_tools = [
    {"name": "ask_user", "description": "Ask the user a question", "parameters": {"type": "object"}},
]

result = await runner.run(
    {"name": "clarifier", "instructions": "Ask the user for clarification when needed."},
    "Help me plan a trip",
    tool_handlers=tool_handlers,
    client_tools=client_tools,
)