Skip to main content
When you call execute_agent or run_agent in Python, the client uses a streaming integration by default: after the run starts, it keeps a long-lived HTTP connection open so the execution API can push task updates as they happen. In browser and HTTP terms this pattern is usually implemented with server-sent events (SSE) — one directional stream from server to client over a regular HTTP response. That is a good fit when your caller is fine holding the connection open for the whole run (interactive scripts, app servers with generous timeouts). In practice the stream can still be cut short—idle timeouts, proxies, load balancers, deploys, or network blips—so production integrations should include retry or resume logic: for example restart the wait using the same task_id, or switch to webhook delivery if you cannot tolerate disconnects. That makes this the preferred way to build scripts and services that should await a full run in a single async function, without running your own status polling loop.

​
What the client does

  1. POSTs to start the workflow run and receives a task_id.
  2. Subscribes to the run event stream on that connection.
  3. On sync events, if the run is paused for tools, invokes your app-defined tools and POSTs tool responses.
  4. Repeats until the run completes, fails, or is cancelled.
For multi-day runs or architectures where you cannot hold a connection open, use webhooks instead — see Running agents.

​
Register handlers and run

Tool handlers are async functions: (args: dict, agent_id: str, task_id: int) -> str. The return value must be a string (often JSON text). The tool name must match a tool registered for the agent on the server. You may pass None for the in-memory schema slot when you only supply the handler; the platform still needs the tool definition (see Register server tools). 
async def get_current_time(args: dict, agent_id: str, task_id: int) -> str:
    from datetime import datetime, timezone
    return datetime.now(timezone.utc).isoformat()

client.set_tools({
    "get_current_time": (None, get_current_time),
})

final = await client.run_agent(
    agent_id="YOUR_AGENT_UUID",
    string_inputs={"1": "What time is it?"},
    environment="playground",
)
The same handler shape is used from webhook code via process_webhook_tools; see Webhooks.