Skip to main content

​
Using the Python SDK

The AgentFlow Python SDK exposes AgentFlow for synchronous code and AsyncAgentFlow for async applications.
1

Install the SDK

uv add agentflow --default-index https://<package-index>/simple/
# or
python -m pip install --index-url https://<package-index>/simple/ agentflow
2

Authenticate

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>
3

Chat with an agent

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 meetings do I have this week?")
    print(response.text)
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.
For local backend development, run agentflow login --method dev --endpoint http://localhost:8001 --profile local after starting the stack with DEV_AUTH_BYPASS=true. The orchestrator automatically delegates your request to the right sub-agent - in this case, MeetingsAgent - which uses its calendar tools to fetch and return your schedule.

​
Streaming responses

For real-time UIs, stream typed events as they arrive: 
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="Draft an email to the Acme team about the Q3 review",
        ):
            if isinstance(event, TextDelta):
                print(event.text, end="", flush=True)

asyncio.run(main())

​
Structured output

Parse final responses directly into Pydantic models: 
from agentflow import AgentFlow, RunOptions
from pydantic import BaseModel

class SentimentAnalysis(BaseModel):
    sentiment: str
    confidence: float
    reasoning: str

with AgentFlow.from_profile("prod") as client:
    agent_id = client.agents.by_name("MainAgent").id

    result = client.agents.run(
        agent_id=agent_id,
        message="Analyze the sentiment of this customer email: ...",
        options=RunOptions(response_model=SentimentAnalysis),
    )

print(result.structured.sentiment)

​
Using the REST API

Every SDK operation maps to a REST endpoint. Customer-facing endpoints are mounted under /api/v1.

​
Discover agents

curl -H "Authorization: Bearer $TOKEN" \
  https://your-instance.agentflow.ai/api/v1/agents

[
  {
    "id": "agent_abc123",
    "name": "MainAgent",
    "description": "Primary AI assistant that coordinates with specialized sub-agents"
  }
]

​
Send a message (non-streaming)

curl -X POST \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  https://your-instance.agentflow.ai/api/v1/agent/agent_abc123/chat \
  -d '{
    "message": "What open opportunities do we have with Acme Corp?",
    "conversation_id": "conv_001",
    "message_id": "msg_001",
    "stream": false,
    "model": "openai/gpt-5.4-mini",
    "temperature": 1.0,
    "knowledge_bases": ["550e8400-e29b-41d4-a716-446655440000"],
    "context_refs": [
      { "system": "crm", "type": "account", "id": "001ABC123" }
    ]
  }'

​
Stream a response (SSE)

curl -N -X POST \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  https://your-instance.agentflow.ai/api/v1/agent/agent_abc123/chat \
  -d '{
    "message": "Research recent news about Acme Corp",
    "conversation_id": "conv_002",
    "message_id": "msg_002",
    "stream": true
  }'

id: 0
data: {"type":"start","call_id":"call_1","content":{"name":"MainAgent"},"seq":0}
id: 1
data: {"type":"delta","call_id":"call_1","content":"Let me research that...","seq":1}
id: 12
data: {"type":"end","call_id":"call_1","seq":12}

​
Interactive API docs

When the server is running, full interactive documentation is available:
  • Swagger UI: GET /docs
  • ReDoc: GET /redoc
  • OpenAPI spec: GET /openapi.json

​
Next steps

Architecture

Understand how multi-agent orchestration works

SDK Reference

Full Python SDK documentation