Agents

Agents are accessed through client.agents. Methods that take agent_id expect the agent UUID used by /api/v1/agent/{agent_id}/...; resolve names once with client.agents.list().

Retrieving agents

from agentflow import AsyncAgentFlow

async with AsyncAgentFlow.from_profile("local") as client:
    agents = await client.agents.list()
    agent_id = {agent.name: agent.id for agent in agents}["MainAgent"]

    agent = await client.agents.retrieve(agent_id)
    config = await client.agents.get_config(agent_id)

Updating agents

updated = await client.agents.update(
    agent_id,
    system_prompt="Updated instructions...",
    enable_planning=True,
    llm_config={"model": "openai/gpt-5.4-mini", "temperature": 1.0, "max_tokens": 4000},
)

Running agents

Simple response

result = await client.agents.run(agent_id=agent_id, message="What's our revenue trend?")
print(result.text)

Streaming

from agentflow.events import ArtifactStatus, FinalResponse, TextDelta, ToolProgress, ToolResultDelta

async for event in client.agents.stream(agent_id=agent_id, message="Analyze Q3 pipeline"):
    if isinstance(event, TextDelta):
        print(event.text, end="", flush=True)
    elif isinstance(event, ToolProgress):
        print(f"\nTool progress: {event.content}")
    elif isinstance(event, ToolResultDelta):
        print(f"\nTool result update: {event.content}")
    elif isinstance(event, ArtifactStatus) and event.type == "artifact_completed":
        print(f"\nArtifact ready: {event.artifact_id}")
    elif isinstance(event, FinalResponse):
        print(event.text)

Every typed event includes the original backend event dictionary on event.raw; use that when an agent integration needs the raw SSE payload fields while still consuming the supported typed stream.

Structured output

from agentflow import RunOptions
from pydantic import BaseModel

class Forecast(BaseModel):
    projected_revenue: float
    confidence: str
    key_risks: list[str]

forecast = await client.agents.run(
    agent_id=agent_id,
    message="Forecast Q4 revenue based on current pipeline",
    options=RunOptions(response_model=Forecast),
)

print(forecast.structured.projected_revenue)

With conversation context

conversation = await client.conversations.create(agent_id=agent_id)

await conversation.send("Show me the pipeline")
await conversation.send("Filter to deals over $100K")
result = await conversation.send("What's the total value?")

With run options

Pass current chat request fields directly on RunOptions.

from agentflow import InlinePromptBlock, RetrievalOptions, RunOptions

result = await client.agents.run(
    agent_id=agent_id,
    message="Analyze Acme and recommend next steps",
    options=RunOptions(
        verbose=True,
        model="openai/gpt-5",
        temperature=0.2,
        max_tokens=4000,
        knowledge_bases=["kb_product_docs"],
        retrieval_options=RetrievalOptions(top_k=8, search_type="hybrid"),
        context_refs=[{"system": "crm", "type": "account", "id": "001ABC123"}],
        prompt_blocks=[
            InlinePromptBlock(
                name="selected_account",
                body="Acme Corp, ARR $120k, renewal in 31 days",
                tags=["selection", "context"],
            )
        ],
    ),
)

Sub-agent management

support = await client.agents.subagents.create(
    agent_id,
    name="TierTwoSupport",
    description="Handles escalated support tickets",
    system_prompt="You handle escalated support tickets.",
    enable_planning=True,
    knowledge_bases=["550e8400-e29b-41d4-a716-446655440000"],
    tool_assignments=["lookup_ticket"],
)

sub_agents = await client.agents.subagents.list(agent_id)

knowledge_bases are KB IDs. tool_assignments on create/update are tool names; the explicit assign/remove helpers use a tool UUID.

Direct sub-agent run

from agentflow import RunOptions

research = next(agent for agent in sub_agents if agent.name == "ResearchAgent")

result = await client.agents.subagents.run(
    agent_id,
    research.id,
    message="Research Acme's latest product announcements",
    conversation_id="conv_research_001",
    message_id="msg_research_001",
    options=RunOptions(verbose=True),
)
print(result.text)

Direct sub-agent stream

async for event in client.agents.subagents.stream(
    agent_id,
    research.id,
    message="Research Acme's latest product announcements",
    conversation_id="conv_research_001",
    message_id="msg_research_002",
    options=RunOptions(verbose=True),
):
    if event.type == "text_delta":
        print(event.text, end="", flush=True)

Direct sub-agent runs use POST /api/v1/agent/{agent_id}/subagents/{subagent_id}/run. They execute the specialist directly as the root run. Delegated sub-agent calls still occur when the parent agent chooses a sub-agent tool during normal chat.

Tool management

tools = await client.agents.tools.list(agent_id)

tool = await client.agents.tools.create(
    agent_id,
    name="lookup_order",
    description="Look up order details from the fulfillment system",
    function_code="async def lookup_order(order_id: str) -> dict:\n    return {'order_id': order_id}",
    require_approval=True,
)

await client.agents.tools.set_approval(agent_id, tool.id, enabled=True)
await client.agents.tools.delete(agent_id, tool.id)

Use client.agents.tools.delete(agent_id, tool_id_or_name) to remove a tool from an agent. For sub-agent assignments, use client.agents.subagents.unassign_tool(agent_id, subagent_id, tool=tool_id_or_name) so the shared tool definition remains available elsewhere.

Questions

When a run emits question_required, respond with the request ID and answers keyed by question item ID:

async for event in client.agents.stream(agent_id=agent_id, message="Create the follow-up task"):
    if event.type == "question_required":
        await client.questions.respond(
            event.question_id,
            answers={"assignee": "user_123", "priority": "high"},
        )

For multiple-choice questions, use the emitted option IDs. questions.respond maps to POST /api/v1/questions/{question_id}/respond. Questions are not a background-task API; they pause the active run until answered or timed out.

Background tasks

Non-blocking tool or sub-agent calls return task handles in the final response metrics and as model-visible tool results:

result = await client.agents.run(
    agent_id=agent_id,
    message="Start a long account research task in the background",
    options=RunOptions(verbose=True),
)

final_event = next(event for event in reversed(result.events) if event.type == "final_response")
custom_metrics = final_event.metrics.get("custom_metrics", {})
for task in custom_metrics.get("background_tasks", []):
    status_events = await client.agents.tools.run(
        agent_id,
        "get_background_sub_agent_result",
        arguments={"task_id": task["task_id"], "include_output": True},
        conversation_id="conv_status_001",
        message_id="msg_status_001",
    )
    print(status_events[-1])

Cancel queued or running background jobs with client.background.cancel(task_id). Foreground runs can be cancelled by call_id, by server-side run ID, or by the original conversation/message pair:

await client.cancellations.cancel("call_abc123")
await client.cancellations.cancel_run("run_abc123")
await client.cancellations.cancel_conversation_message(
    conversation_id="conv_001",
    message_id="msg_001",
)

Background sub-agent helper tools available to the orchestrator are get_background_sub_agent_result, fetch_sub_agent_progress, and message_sub_agent. Use fetch_sub_agent_progress for a non-interfering LLM progress answer, and message_sub_agent with mode="send" or mode="send_and_wait" when the parent needs to send an actual message into the child run.

Knowledge base management

await client.agents.knowledge_bases.assign(agent_id, "kb_product_docs")
kbs = await client.agents.knowledge_bases.list(agent_id)

response = await client.agents.knowledge_bases.search(
    agent_id,
    kbs[0].id,
    query="How does pricing work?",
    limit=5,
)

for document in response.documents:
    print(document.content[:200])

await client.agents.knowledge_bases.unassign(agent_id, "kb_product_docs")

Prompt blocks

blocks = await client.prompt_blocks.list()

await client.prompt_blocks.create(
    name="account_guidance",
    description="Guidance for account analysis.",
    message="system",
    type="static",
    mode="cached",
    scope="conversation",
    tags=["crm", "guidance"],
    body="Prioritize renewal risk and expansion signals.",
)

preview = await client.prompt_blocks.preview("account_guidance")

Context blocks

blocks = await client.agents.list_context_blocks(agent_id)

rendered = await client.agents.build_context(agent_id, include_content=True)
for name, content in rendered.content_by_name.items():
    print(f"--- {name} ---")
    print(content[:200])

await client.agents.preload_context(agent_id)

Properties

Property	Type	Description
`id`	`str`	Agent UUID
`name`	`str`	Agent name
`description`	`str	None`	Agent description
`system_prompt`	`str	None`	System prompt
`llm_config`	`dict`	Model configuration
`knowledge_bases`	`list[dict]`	Knowledge bases assigned to the agent

​Agents

​Retrieving agents

​Updating agents

​Running agents

​Simple response

​Streaming

​Structured output

​With conversation context

​With run options

​Sub-agent management

​Direct sub-agent run

​Direct sub-agent stream

​Tool management

​Questions

​Background tasks

​Knowledge base management

​Prompt blocks

​Context blocks

​Properties