Skip to main content
Tools have two parts that must stay aligned:
  • Server registry: JSON schema metadata so models know tool names and arguments.
  • App-defined tool: your code that runs when the model asks to call that tool.
This guide covers both the registration step and the runtime handler contract for SSE and webhook flows.

​
Register one tool

This example defines a parse_pdf tool with a single required string argument file_path.
Python
from asapi.models.tools import ToolDefinition, ParametersSchema, ParameterProperty

definitions = [
    ToolDefinition(
        name="parse_pdf",
        namespace="main",
        description="Extract text from a PDF given file_path.",
        schema_data=ParametersSchema(
            properties={
                "file_path": ParameterProperty(
                    type="string",
                    description="Reference from run-input store or path.",
                ),
            },
            required=["file_path"],
        ),
    ),
]

await client.tools.register_tools(definitions)

​
List tools

Python
page = await client.tools.get_tools(namespace="main", limit=50)
Use listing to confirm registration or to debug missing-tool errors during runs.

​
App-defined tool contract

For Python integrations, handlers use this signature:
  • Signature: async def tool(args: dict, agent_id: str, task_id: int) -> str
  • Return: a string. For structured output, return JSON text (for example json.dumps({...})).
Why all three inputs matter:
  • args: model-supplied arguments for that tool call.
  • agent_id: which workflow invoked the tool (lets one handler support multiple agents).
  • task_id: the specific run id for logging, tracing, and run-scoped lookups.

​
SSE runtime path

For streaming runs, register handlers with set_tools before execute_agent / run_agent. The client invokes them automatically when the run pauses for tool calls. 
import json

async def parse_pdf(args: dict, agent_id: str, task_id: int) -> str:
    path = (args or {}).get("file_path", "")
    # ... read the file reference and extract text ...
    return json.dumps({"text": "..."})

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

​
Webhook runtime path

If you run with webhooks, parse webhook payloads and submit tool responses explicitly:
  1. tool_results = await client.process_webhook_tools(request, tools_map)
  2. await client.send_tool_responses(request.agent_id, request.task_id, tool_results.results)
Minimal sketch: 
from asapi import AgentGraphRunStatusResponse

async def handle_agent_webhook(body: dict, client):
    request = AgentGraphRunStatusResponse.model_validate(body)
    if request.status != "awaiting_tool_calls":
        return {"ok": True}

    tool_results = await client.process_webhook_tools(request, {
        "parse_pdf": (None, parse_pdf),
    })
    await client.send_tool_responses(
        request.agent_id,
        request.task_id,
        tool_results.results,
    )
    return {"ok": True, "status": "tool_responses_sent"}
Do not use polling loops to discover tool calls; use SSE or webhooks as documented.

​
See also