Tools & Composability

Tools are the atomic units of agent capability. Every action an agent takes, from querying a database to sending an email, is a tool invocation. AgentFlow’s tool system provides a global registry, decorator-based registration, SDK registration, approval gates, persisted assignments, and shared instances across agents.

Registering tools

Server-side decorators

Tools checked into the AgentFlow server are defined with the @tool decorator and discovered at startup. Import the decorator from framework.decorators:

from framework.decorators import tool

@tool(
    name="search_accounts",
    description="Search CRM accounts by name, industry, or revenue range",
    agent="RecordsAgent",
    tags=["crm", "search"],
    group="CRM Tools",
    require_approval=False,
)
async def search_accounts(
    query: str,
    industry: str | None = None,
    min_revenue: float | None = None,
    limit: int = 10,
) -> list[dict]:
    """Search and filter CRM accounts."""
    ...

Use agent="RecordsAgent" for one target agent, or pass a list to share the same code-defined tool across agents:

@tool(
    name="search_accounts",
    description="Search CRM accounts by name, industry, or revenue range",
    agent=["RecordsAgent", "MainAgent"],
)
async def search_accounts(query: str, limit: int = 10) -> list[dict]:
    ...

The keyword is intentionally singular: agent= accepts str | list[str] | None.

SDK registration

Runtime tools are registered through the agent-scoped SDK resource. Use client.agents.tools.register(...) for a local Python callable, or pass function_code to client.agents.tools.create(...) when you want to send source explicitly:

from agentflow import AsyncAgentFlow

async with AsyncAgentFlow.from_profile("local") as client:
    tool = await client.agents.tools.create(
        agent_id,
        name="calculate_roi",
        description="Calculate return on investment with annualized rate.",
        function_code="""
async def calculate_roi(investment: float, returns: float, period_months: int) -> dict:
    roi = (returns - investment) / investment
    annualized = (1 + roi) ** (12 / period_months) - 1
    return {
        "roi_percentage": round(roi * 100, 2),
        "annualized_rate": round(annualized * 100, 2),
    }
""",
        args_schema={
            "type": "object",
            "properties": {
                "investment": {"type": "number"},
                "returns": {"type": "number"},
                "period_months": {"type": "integer"},
            },
            "required": ["investment", "returns", "period_months"],
        },
        tags=["finance", "analytics"],
        require_approval=False,
    )

function_code may define either a sync or async function; async functions are awaited by the runtime:

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": []}
""",
    tags=["data", "sql"],
    require_approval=True,
    execution_timeout=30.0,
)

There is no public agent.register_tool(...) SDK method. Use client.agents.tools.register(...) for local callables, client.agents.tools.create(...) for source-backed runtime tools, or from framework.decorators import tool for server-side source files.

Registry and identifiers

The tool registry is a singleton that deduplicates tool instances across agents:

Tools with the same name share one persisted definition and stable ID.
Agent and sub-agent assignments are durable and tenant-scoped.
Agent targeting via agent="AgentName" or agent=["AgentA", "AgentB"] binds decorator tools to one or more agents.
Runtime registration can assign the same tool to additional agents or a target sub-agent.

Use tool.id as the canonical persisted identifier. Use tool.name as the model-facing callable name. API path parameters named tool_identifier accept either the ID or the name; direct execution path parameters named tool_id accept either form.

Tool execution

Direct execution

Tools can be run independently, outside of an agent conversation:

POST /api/v1/agent/{agent_id}/tools/{tool_id}/run

{
  "arguments": {
    "query": "Acme",
    "limit": 5
  },
  "conversation_id": "conv_acme_search",
  "message_id": "msg_acme_search",
  "idempotency_key": "toolrun_acme_search_001",
  "stream": true
}

Raw REST requests must include conversation_id and message_id so events, artifacts, and cache refs have execution context. SDK helpers mint IDs when omitted, but production integrations should pass stable IDs from their own message model. Pass idempotency_key to reject duplicate in-flight submissions for a side-effecting run. With stream: true, the response is an SSE stream of execution events. With stream: false, the same event objects are collected into a JSON response under events. The SDK provides collected and streaming helpers:

events = await client.agents.tools.run(
    agent_id,
    "search_accounts",
    arguments={"query": "Acme", "limit": 5},
    conversation_id="conv_acme_search",
    message_id="msg_acme_search",
    idempotency_key="toolrun_acme_search_001",
)

async for event in client.agents.tools.stream_raw(
    agent_id,
    "search_accounts",
    arguments={"query": "Acme", "limit": 5},
    conversation_id="conv_acme_search",
    message_id="msg_acme_search",
):
    print(event)

Use stream_raw() when you want the backend SSE payloads exactly as emitted. Use stream() when you want the SDK’s typed event projection.

Within agent flow

When an agent decides to use a tool during a conversation, the execution is part of the event stream:

id: 3
data: {"type":"start","call_id":"tool_1","parent_call_id":"agent_1","content":{"name":"search_accounts"},"seq":3}
id: 4
data: {"type":"end","call_id":"tool_1","content":{"result":[...]},"seq":4}

Approval gates

Any tool can require human approval before execution. When require_approval=True, the agent pauses, emits an approval request event, and waits for a user decision:

tool = await client.agents.tools.create(
    agent_id,
    name="send_contract",
    description="Send a contract for signature.",
    function_code="""
async def send_contract(account_id: str, recipient: str) -> dict:
    return {"status": "queued", "account_id": account_id, "recipient": recipient}
""",
    require_approval=True,
)

The approval flow:

Agent decides to call send_contract
Execution pauses; an approval request is created
User reviews the tool arguments via the Approvals API
User approves, approves with edited arguments, or denies
If approved, tool executes; if denied, agent is informed and can adjust

Approval API

# List pending approvals
GET /api/v1/approvals/pending

# Review a specific approval
GET /api/v1/approvals/{approval_id}

# Approve or deny, optionally with reviewed arguments
POST /api/v1/approvals/{approval_id}/respond
{
  "status": "approved",
  "reason": "Verified the contract details",
  "modified_arguments": {
    "recipient": "[email protected]"
  }
}

# Bulk approve/deny
POST /api/v1/approvals/bulk-respond

# Approval statistics per tool
GET /api/v1/approvals/tools/{tool_id}/stats

# Dashboard summary
GET /api/v1/approvals/dashboard/summary

modified_arguments is optional. When present on an approved response, the tool executes with the reviewed arguments instead of the originally proposed arguments.

Lifecycle and permissions

Tool registration, updates, deletes, and assignments are persisted. A server restart should not resurrect a removed assignment or lose an updated runtime tool.

Operation	Behavior
Register	Creates or reuses the persisted tool definition, then assigns it to the target agent or sub-agent
Update	Writes a new persisted version of the definition, schema, metadata, approval config, or credentials
Unassign	Removes the agent or sub-agent assignment while leaving the tool available elsewhere
Delete	Removes the persisted tool definition and its assignments when the caller requests a durable delete

Updating or removing a tool invalidates the tenant tool-definition cache so future agent runs use the current definition. In the Python SDK, agent-level removal is client.agents.tools.delete(agent_id, tool_id_or_name). Sub-agent assignment removal is client.agents.subagents.unassign_tool(agent_id, subagent_id, tool=tool_id_or_name). All tool management and direct execution endpoints require an AgentFlow bearer token. Management calls require tool administration permissions for the tenant and agent. Direct execution requires permission to run the target agent’s tools and still enforces approval gates. Downstream integrations run with the request principal’s context plus any credentials configured on the tool.

Built-in tool categories

AgentFlow ships with tools across multiple integration domains:

Category	Tool count	Examples
CRM / Records	18+	Account search, opportunity CRUD, contact management, pipeline queries
Email	10+	Draft, send, reply, search, thread management, inbox operations
Meetings	8+	Schedule, availability check, transcript analysis, agenda prep
Research	6+	Web search, company news, SEC filings, industry reports
Tasks	6+	Create, assign, track, prioritize, due date management
Data	5+	SQL queries, report generation, data visualization
Knowledge	4+	KB search, document retrieval, context injection
Internal	4+	Planning, retrieval, reasoning, reflection (framework capabilities)

Result caching and summarizers

Tools that return large structured results (CRM queries, email threads, meeting lists) use the artifact-backed result cache to stay within LLM context limits. The cache summarizer is registered next to the artifact type, not on the @tool decorator:

from framework.artifacts.specs import artifact
from framework.tools.result_summarizers import register_cached_tool


@artifact(type="account_health", label="Account Health", display_mode="panel")
def summarize_account_health(content: dict) -> dict:
    return {
        "title": content.get("account_name", "Account health"),
        "score": content.get("health_score"),
        "risk_factors": len(content.get("risks", [])),
    }


register_cached_tool(tool_name="get_account_health", summarizer=summarize_account_health)

When register_cached_tool(...) is paired with the sibling @artifact(...) summarizer:

The full result is stored in a TTL-scoped result cache.
The summarizer produces a compact dict, typically about 200 tokens instead of 10,000+.
The summary is returned to the LLM with a cache_id reference.
A matching artifact type is auto-registered so the UI can render the data as a panel.
The full data remains available through the conversation-scoped cached-results API while the entry is valid.

This means agents can work with large datasets without context window pressure. See Artifacts for more detail on the summarizer system. When a tool result summary includes a cache_id, clients can retrieve the full cached payload through the SDK instead of asking the model to reproduce the data:

full_result = await client.cached_results.retrieve(
    "conv_001",
    cache_id="cr_abc123",
)

Cache entries are scoped to the conversation and expire according to the tool’s cache policy. Re-run the source tool when a cache ref is expired or from an unrelated conversation.

Tool management API

# List tools for an agent
GET /api/v1/agent/{agent_id}/tools

# Get tool details
GET /api/v1/agent/{agent_id}/tools/{tool_identifier}

# Register a new tool
POST /api/v1/agent/{agent_id}/tools

# Update a persisted tool definition or metadata
PATCH /api/v1/agent/{agent_id}/tools/{tool_identifier}

# Execute a tool directly
POST /api/v1/agent/{agent_id}/tools/{tool_id}/run

# Configure approval settings
PUT /api/v1/agent/{agent_id}/tools/{tool_identifier}/approval

# Durably unassign or delete a tool
DELETE /api/v1/agent/{agent_id}/tools/{tool_identifier}

​Tools & Composability

​Registering tools

​Server-side decorators

​SDK registration

​Registry and identifiers

​Tool execution

​Direct execution

​Within agent flow

​Approval gates

​Approval API

​Lifecycle and permissions

​Built-in tool categories

​Result caching and summarizers

​Tool management API