Skip to main content

​
Python SDK

The AgentFlow Python SDK exposes a single namespaced root client: AgentFlow for synchronous code and AsyncAgentFlow for async applications. One client represents one tenant and one principal.

​
Installation

uv add agentflow --default-index https://<package-index>/simple/
# or
python -m pip install --index-url https://<package-index>/simple/ agentflow
The public PyPI package named agentflow is not the private AgentFlow SDK unless your organization has published it there. Use your onboarding package source, private package index, or internal wheel.

​
Authenticate

Create a profile with the CLI: 
agentflow login \
  --endpoint https://your-instance.agentflow.ai \
  --profile prod \
  --method device \
  --client-id <auth0-client-id> \
  --audience <auth0-api-audience> \
  --auth0-domain <auth0-domain>
Use --method m2m for service clients and CI. Use --method dev only against a local backend started with DEV_AUTH_BYPASS=true.

​
Quick start

from agentflow import AgentFlow

with AgentFlow.from_profile("prod") as client:
    agent_id = client.agents.by_name("MainAgent").id
    result = client.agents.run(agent_id=agent_id, message="What meetings do I have this week?")
    print(result.text)
Async mirror: 
from agentflow import AsyncAgentFlow

async with AsyncAgentFlow.from_profile("prod") as client:
    agent_id = (await client.agents.by_name("MainAgent")).id
    result = await client.agents.run(agent_id=agent_id, message="What meetings do I have this week?")
    print(result.text)

​
Core resources

The SDK is organized around resources that map to /api/v1 backend routes:
ResourcePurposeExample
client.agentsAgent CRUD, runs, context blocksclient.agents.run(...)
client.agents.toolsTools scoped to an agentclient.agents.tools.create(agent_id, ...)
client.agents.subagentsDynamic sub-agent managementclient.agents.subagents.create(parent_id, ...)
client.agents.knowledge_basesAgent-scoped KB assignment and searchclient.agents.knowledge_bases.search(...)
client.prompt_blocksPrompt block management and previewsclient.prompt_blocks.create(...)
client.conversationsMulti-turn conversation objectsclient.conversations.create(agent_id=agent_id)
client.memoryPersistent core and archival memoryclient.memory.recall("Acme renewal")
client.completionsSync classification, extraction, summarization, enrichment, and custom runsclient.completions.classify(...)
client.cached_resultsConversation-scoped cached tool artifactsclient.cached_results.retrieve(conversation_id, cache_id)
client.knowledge_basesGlobal KB management and searchclient.knowledge_bases.list()
client.approvalsTool approval workflowsclient.approvals.list()
client.artifactsArtifact retrieval and actionsclient.artifacts.retrieve(artifact_id)
client.batchesAsync batch jobs, file jobs, polling, and output downloadsclient.batches.submit(...)
client.attachmentsFile upload/downloadclient.attachments.upload(...)
client.backgroundInspect and cancel background tasksclient.background.cancel(task_id)
client.cancellationsCancel active calls, run IDs, and conversation/message pairsclient.cancellations.cancel_run(run_id)
client.modelsModel catalog and BYO LLM admin policyclient.models.list()
client.systemHealth and metadataclient.system.health()

​
Explicit credentials

Profiles are the preferred path. If you already have an AgentFlow bearer token, pass it directly: 
from agentflow import AgentFlow

client = AgentFlow(
    endpoint="https://your-instance.agentflow.ai",
    access_token="<agentflow-access-token>",
)

export AGENTFLOW_ENDPOINT=https://your-instance.agentflow.ai
export AGENTFLOW_ACCESS_TOKEN=<agentflow-access-token>
Machine clients can use client credentials with delegated-user headers. Request the scopes for the user-owned APIs you need: 
from agentflow import AgentFlow

client = AgentFlow(
    endpoint="https://your-instance.agentflow.ai",
    client_id="<auth0-client-id>",
    client_secret="<auth0-client-secret>",
    tenant="company-a",
    on_behalf_of="svc-bot",
    scope="agentflow:conversations agentflow:cancellations",
    audience="<auth0-api-audience>",
    auth0_domain="<auth0-domain>",
)

​
Running agents

from agentflow import AsyncAgentFlow, InlinePromptBlock, MemoryPolicy, RetrievalOptions, RunOptions

async with AsyncAgentFlow.from_profile("local") as client:
    agent_id = (await client.agents.by_name("MainAgent")).id

    result = await client.agents.run(
        agent_id=agent_id,
        message="Summarize the Acme account",
        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"),
            memory_policy=MemoryPolicy(archival_mode="auto", archival_limit=5),
            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"],
                )
            ],
        ),
    )
    print(result.text)
RunOptions maps typed fields directly to POST /api/v1/agent/{agent_id}/chat, including knowledge base, bounded retrieval_options, per-run memory_policy, model, context refs, inline prompt_blocks, and streaming controls.

​
Streaming

from agentflow.events import FinalResponse, TextDelta

async with AsyncAgentFlow.from_profile("local") as client:
    agent_id = (await client.agents.by_name("MainAgent")).id

    async for event in client.agents.stream(agent_id=agent_id, message="Analyze our Q3 pipeline"):
        if isinstance(event, TextDelta):
            print(event.text, end="", flush=True)
        elif isinstance(event, FinalResponse):
            print(event.text)
Each typed event includes the original backend SSE payload on event.raw. For agent runs, this is the supported raw event view in the SDK. 
async for event in client.agents.stream(agent_id=agent_id, message="Analyze our Q3 pipeline"):
    print(event.type, event.raw)
Mutating SDK calls accept an idempotency_key option where the API supports request de-duplication. For direct tool runs, the key rejects duplicate in-flight submissions; it is not a persisted result replay key. For chat, pass a stable message_id when retrying the same user action.

​
Conversations

For multi-turn work, create a conversation and reuse it: 
async with AsyncAgentFlow.from_profile("local") as client:
    agent_id = (await client.agents.by_name("MainAgent")).id
    conversation = await client.conversations.create(agent_id=agent_id)

    first = await conversation.send("Show me the pipeline")
    second = await conversation.send("Now filter to deals over $100K")