Skip to main content
The Python SDK provides sync and async clients for the AgentFlow API.

​
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://api.example.com \
  --profile prod \
  --method device \
  --client-id <auth0-client-id> \
  --audience <auth0-api-audience> \
  --auth0-domain <auth0-domain>
Use --method m2m for service clients and --method dev for local backends started with DEV_AUTH_BYPASS=true.

​
Initialize from a profile

from agentflow import AgentFlow

client = AgentFlow.from_profile("prod")
If you already have a bearer token, pass it directly or set environment variables: 
from agentflow import AgentFlow

client = AgentFlow(endpoint="https://api.example.com", access_token="<agentflow-access-token>")

export AGENTFLOW_ENDPOINT=https://api.example.com
export AGENTFLOW_ACCESS_TOKEN=<agentflow-access-token>
For machine-to-machine clients that act on behalf of a user, pass scope="agentflow:conversations agentflow:cancellations" when constructing the client, set AGENTFLOW_SCOPE for environment-configured clients, or add the same scope to a saved M2M profile.

​
Agents

​
Chat

from agentflow import AgentFlow

with AgentFlow.from_profile("prod") as client:
    agent_id = client.agents.by_name("MainAgent").id
    response = client.agents.run(agent_id=agent_id, message="What can you do?")
    print(response.text)

​
Stream

import asyncio
from agentflow import AsyncAgentFlow
from agentflow.events import TextDelta

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

        async for event in client.agents.stream(agent_id=agent_id, message="Hello"):
            if isinstance(event, TextDelta):
                print(event.text, end="", flush=True)

asyncio.run(main())

​
Chat with conversation context

from agentflow import AsyncAgentFlow

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

    await conversation.send("Show me the pipeline")
    response = await conversation.send("Follow up on the last topic")

​
Knowledge Bases

from agentflow import AsyncAgentFlow

async with AsyncAgentFlow.from_profile("local") as client:
    results = await client.knowledge_bases.search(
        "550e8400-e29b-41d4-a716-446655440000",
        query="How does billing work?",
        limit=5,
    )

    for result in results.documents:
        print(result.content)

​
Completions and Batches

Use client.completions for sync inline completion helpers and client.batches for async batch jobs. 
import time

from agentflow import AgentFlow

with AgentFlow.from_profile("prod") as client:
    labels = client.completions.classify(
        items=[{"content": "Great product, slow delivery."}],
        categories=["positive", "negative", "mixed"],
    )
    print(labels.results[0]["result"])

    batch = client.batches.submit(
        operation="extract",
        items=[{"content": "Sarah Chen, VP Engineering at Acme Corp, wants 500 seats by Q3."}],
        extraction_instructions="Extract contact, title, company, seat_count, and timeline.",
    )
    print(batch.id, batch.status)

    while batch.status not in {"completed", "failed", "cancelled"}:
        time.sleep(2)
        batch = client.batches.retrieve(batch.id)

    page = client.batches.output(batch.id, format="json", offset=0, limit=1000)
    for row in page.results:
        print(row["result"])

​
BYO LLM Admin

Admins can manage encrypted tenant provider keys and model policy through client.models. 
from agentflow import AgentFlow

with AgentFlow.from_profile("admin") as client:
    options = client.models.model_options()
    print(options.use_cases)

    client.models.save_configuration(
        provider="anthropic",
        api_key="sk-ant-...",
        allowed_models=[
            "anthropic/claude-haiku-4-5-20251101",
            "anthropic/claude-sonnet-4-5-20250929",
        ],
        default_models={
            "chat": "anthropic/claude-sonnet-4-5-20250929",
            "tool": "anthropic/claude-haiku-4-5-20251101",
            "title": "anthropic/claude-haiku-4-5-20251101",
            "enrichment": "anthropic/claude-haiku-4-5-20251101",
        },
    )

    client.models.save_configuration(
        provider="google",
        api_key="AIza...",
        allowed_models=["google/gemini-2.5-flash"],
        default_models={
            "follow_up_questions": "google/gemini-2.5-flash",
            "autocomplete": "google/gemini-2.5-flash",
        },
    )

    policy = client.models.configurations()
    print(policy.mode, policy.allowed_models)
Secrets are never returned by the SDK. Provider configurations expose only metadata such as key_last_four, allowed_models, and default_models.

​
Conversations

​
List

conversations = client.conversations.list()

​
Search

matches = client.conversations.search("Acme renewal")

​
Get messages

conversation = client.conversations.retrieve("conv_abc123")
for msg in conversation.messages:
    print(msg.role, msg.content)

​
Memory

from agentflow import MemoryPolicy, RunOptions

core = client.memory.get("core")
client.memory.update(
    "core",
    "- Prefers concise answers",
    expected_updated_at=core.updated_at.isoformat(),
)

for match in client.memory.recall("Acme renewal"):
    print(match.scope, match.content)

result = client.agents.run(
    agent_id="agent_records",
    message="What changed with Acme?",
    options=RunOptions(memory_policy=MemoryPolicy(archival_mode="auto", archival_limit=5)),
)

​
System

​
Health check

status = client.system.health()
print(status)