Skip to main content
Webhook mode fits backends that start a run once, return quickly to the caller, and then learn what happened as the workflow runs: tool rounds, failures, and completion all arrive as out-of-band HTTPS POSTs—you do not keep a streaming connection open and you should not busy-loop the status API.

How webhook mode works

  1. Start the agent — Your service calls the execution API a single time to start the workflow (a “start-only” execute that returns task_id without subscribing to SSE). Store task_id (and any correlation ids you need).
  2. Receive run updates — In Sudoiq, configure your webhook URL. The platform POSTs a JSON body to that URL whenever the run state changes. Each payload should parse to AgentGraphRunStatusResponse (or the equivalent model from asapi). Inspect status for the current phase (for example completed, terminal failure, or awaiting_tool_calls).
  3. Handle tool calls — When status is awaiting_tool_calls, the model has requested tools your process must execute. Run your handlers, then call send_tool_responses with the same agent_id and task_id so the run can continue; you may receive another webhook on the next pause or when the run finishes.
Do not poll get_agent_task_status in a loop; rely on deliveries to your webhook (retries and idempotency are covered under Operational notes).

Start a run (then wait for callbacks)

Use the SDK method that starts a workflow run and returns task_id without opening a streaming connection. The exact symbol ships with your asapi release (often a “start only” execute variant).
Python
started = await client.start_agent_run(
    agent_id="YOUR_AGENT_UUID",
    string_inputs={"1": "Hello"},
    environment="playground",
)
# Persist started.task_id; return 202 to your caller. Progress arrives via webhook.
Until your SDK exposes start_agent_run, use the start-only execute method documented in the installed asapi package (same HTTP call as streaming start, without subscribing to the event stream).

Webhook handler: tools and completion

The same process_webhook_tools and send_tool_responses helpers power tool rounds; only delivery differs from the streaming path.
Python
from asapi import AsapiClient, AgentGraphRunStatusResponse

async def handle_agent_webhook(body: dict, client: AsapiClient) -> dict:
    request = AgentGraphRunStatusResponse.model_validate(body)

    if request.status == "completed":
        return {"ok": True, "result": request.result}

    if request.status == "awaiting_tool_calls":
        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()

        tools = {"get_current_time": (None, get_current_time)}
        tool_results = await client.process_webhook_tools(request, tools)
        await client.send_tool_responses(
            request.agent_id,
            request.task_id,
            tool_results.results,
        )
        return {"ok": True, "status": "tool_responses_sent"}

    return {"ok": True, "status": request.status}

Operational notes

  • Persistence — Store task_id, tenant, and correlation ids when you start the run.
  • Idempotency — Providers may retry deliveries; design handlers to tolerate duplicates where possible.
  • Out of scope — These docs do not show while-loop status polling in application code.