Skip to main content

​
Tools

Tools are managed through client.agents.tools because tool assignment is scoped to an agent or sub-agent. Use the persisted tool.id for durable operations and the model-facing tool.name when you want to address a tool by name.

​
Registering tools

Register a Python callable when the tool lives in your application code. The SDK derives the tool name, description, source, and argument schema from the callable: 
async def lookup_order(order_id: str) -> dict:
    """Look up order details from the fulfillment system."""
    return {"order_id": order_id}

tool = await client.agents.tools.register(
    agent_id,
    lookup_order,
    tags=["orders"],
    require_approval=True,
)
Create runtime tools with source code that AgentFlow can persist and execute. function_code may define a sync or async function: 
tool = await client.agents.tools.create(
    agent_id,
    name="query_database",
    description="Execute a read-only SQL query against the analytics database",
    function_code="""
async def query_database(sql: str) -> dict:
    return {"rows": []}
""",
    args_schema={
        "type": "object",
        "properties": {"sql": {"type": "string"}},
        "required": ["sql"],
    },
    tags=["data", "sql"],
    require_approval=True,
    execution_timeout=30.0,
)

​
Retrieving tools

tool = await client.agents.tools.retrieve(agent_id, "query_database")

tools = await client.agents.tools.list(agent_id)
for tool in tools:
    print(f"{tool.id} {tool.name}: {tool.description}")

​
Approval gates

await client.agents.tools.set_approval(agent_id, "query_database", enabled=True, timeout=600)
await client.agents.tools.set_approval(agent_id, "query_database", enabled=False)
Approval responses can include reviewed arguments: 
await client.approvals.respond(
    "appr_abc123",
    decision="approved",
    modified_arguments={"sql": "select * from accounts limit 10"},
)

​
Direct execution

Direct execution runs an already assigned tool. The REST API requires conversation_id and message_id; SDK helpers mint IDs when omitted, but production integrations should pass stable IDs from their own message model. Pass an idempotency_key to reject duplicate in-flight submissions for side-effecting calls. 
from agentflow.events import FinalResponse, ToolCallCompleted

events = await client.agents.tools.run(
    agent_id,
    "query_database",
    arguments={"sql": "select * from accounts limit 10"},
    conversation_id="conv_query_accounts_001",
    message_id="msg_query_accounts_001",
    idempotency_key="query-accounts-001",
)

for event in events:
    if isinstance(event, ToolCallCompleted):
        print(event.result)
    elif isinstance(event, FinalResponse):
        print(event.text)
Stream events as they happen: 
from agentflow.events import ToolCallCompleted

async for event in client.agents.tools.stream(
    agent_id,
    "query_database",
    arguments={"sql": "select * from accounts limit 10"},
    conversation_id="conv_query_accounts_001",
    message_id="msg_query_accounts_001",
):
    if isinstance(event, ToolCallCompleted):
        print(event.result)
Use stream_raw() to receive the backend SSE event dictionaries instead of the SDK’s typed event projection: 
async for event in client.agents.tools.stream_raw(
    agent_id,
    "query_database",
    arguments={"sql": "select * from accounts limit 10"},
    conversation_id="conv_query_accounts_001",
    message_id="msg_query_accounts_001",
):
    print(event["type"], event.get("content"))

​
Dry runs

Use dry runs to test source before registering it: 
result = await client.agents.tools.dry_run(
    agent_id,
    function_code="""
async def analyze_sentiment(text: str) -> dict:
    return {"sentiment": "negative"}
""",
    arguments={"text": "The product quality has been declining lately"},
)
print(result.result)

​
Cached results

Tools with artifact-backed summarizers may return compact summaries with a cache_id. Retrieve the full payload through the SDK while the cache entry is still valid: 
full_result = await client.cached_results.retrieve(
    "conv_001",
    cache_id="cr_abc123",
)

​
Deleting and unassigning tools

Delete a tool when it should no longer be available to the agent: 
await client.agents.tools.delete(agent_id, "query_database")
Agent-level removal uses delete(...), matching the REST DELETE /agent/{agent_id}/tools/{tool_identifier} endpoint. Sub-agent assignment removal uses unassign_tool(...): 
await client.agents.subagents.unassign_tool(
    agent_id,
    subagent_id,
    tool="query_database",
)