FastAPI Integration

Agency Swarm supports serving your agencies and tools as production-ready HTTP APIs using FastAPI. This enables you to interact with your agents and tools over HTTP, integrate with other services, or connect it to web frontends.

Installation

FastAPI integration is an optional installation. To install all required dependencies, run:

pip install "agency-swarm[fastapi]"

Setting Up FastAPI Endpoints

You can expose your agencies and tools as API endpoints using the run_fastapi() function.

Example: Create an API endpoint for a single agency

from agency_swarm import Agency, Agent

agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant."
)

agency = Agency(agent, name="test_agency")

agency.run_fastapi()

run_fastapi parameters

Param	Type	Default	Description
`host`	string	`"0.0.0.0"`	Interface to bind the server.
`port`	integer	`8000`	Port to serve FastAPI on.
`app_token_env`	string	`"APP_TOKEN"`	Env var name for the bearer token; if missing, auth is disabled.
`return_app`	boolean	`false`	Return the FastAPI app instead of running the server.
`cors_origins`	list	`["*"]`	Allowed CORS origins. When set to `["*"]`, `allow_credentials` is automatically set to `False` for security.
`enable_agui`	boolean	`false`	Enable AG-UI streaming (hides the cancel endpoint).
`enable_logging`	boolean	`false`	Track requests and expose `/get_logs` at the server root.
`logs_dir`	string	`"activity-logs"`	Folder for request logs when logging is enabled.
`allowed_local_file_dirs`	list	`None`	Allowlist for local `file_urls`; paths outside the list are rejected.
`enable_realtime`	boolean	`false`	Mount `/<agency_name>/realtime` websocket route(s) in addition to REST endpoints.
`realtime_options`	object	`{}`	Realtime session overrides (for example `provider`, `model`, `voice`, `turn_detection`).
`oauth_user_id_dependency`	FastAPI dependency	`None`	Trusted authentication dependency required for OAuth-enabled agencies. It must return a stable, non-secret user ID.
`verify_oauth_callback_user`	boolean	`false`	Apply `oauth_user_id_dependency` to `/auth/callback` and reject a code whose pending flow belongs to another user. Requires browsers to carry authentication (cookie or session).

Endpoint reference

Agencies are served at:
- /your_agency_name/get_response (POST)
- /your_agency_name/get_response_stream (POST, streaming responses)
- /your_agency_name/cancel_response_stream (POST; not registered when enable_agui=True)
- /your_agency_name/get_metadata (GET)
- /your_agency_name/realtime (WebSocket; when enable_realtime=True)
- /get_logs (GET; when enable_logging=True)
Tools registered via tools=[...] are available at /tool/ToolClassName (BaseTools) or /tool/function_name (function tools).
OpenAPI and interactive docs: /openapi.json, /docs, /redoc.

Response format

Non-streaming (/get_response):

{
  "response": "Hello! How can I help you today?",
  "new_messages": [
    {
      "type": "message",
      "role": "user",
      "content": [{"type": "input_text", "text": "Hello"}],
      "agent": "Assistant",
      "callerAgent": null,
      "timestamp": 1704067200000
    },
    {
      "type": "message",
      "role": "assistant",
      "content": [{"type": "output_text", "text": "Hello! How can I help you today?"}],
      "agent": "Assistant",
      "callerAgent": null,
      "timestamp": 1704067201000
    }
  ],
  "usage": {"input_tokens": 50, "output_tokens": 12, "total_cost": 0.0001},
  "file_ids_map": {"document.pdf": "file-abc123"}
}

Streaming (/get_response_stream):

event: meta
data: {"run_id": "550e8400-e29b-41d4-a716-446655440000"}

data: {"data": {"data": {"type": "response.output_text.delta", "delta": "Hello"}}}
data: {"data": {"data": {"type": "response.output_text.delta", "delta": "!"}}}

event: messages
data: {"new_messages": [...], "run_id": "...", "cancelled": false, "usage": {...}, "file_ids_map": {...}}

event: end
data: [DONE]

Cancel (/cancel_response_stream):

{
  "ok": true,
  "run_id": "550e8400-e29b-41d4-a716-446655440000",
  "cancelled": true,
  "cancel_mode": "immediate",
  "new_messages": [...]
}

Metadata (/get_metadata):

{
  "nodes": [
    {
      "id": "Assistant",
      "type": "agent",
      "data": {
        "label": "Assistant",
        "description": "Main assistant agent",
        "conversationStarters": ["Support: I need help with billing"],
        "quickReplies": ["hi", "hello"],
        "isEntryPoint": true,
        "toolCount": 2,
        "tools": [
          {
            "name": "example_tool",
            "type": "function",
            "description": "...",
            "inputSchema": {"type": "object", "properties": {"text": {"type": "string"}}}
          }
        ]
      }
    }
  ],
  "edges": [
    {"id": "CEO->Developer", "source": "CEO", "target": "Developer", "type": "communication"}
  ],
  "metadata": {
    "agencyName": "my_agency",
    "totalAgents": 2,
    "totalTools": 4,
    "agents": ["CEO", "Developer"],
    "entryPoints": ["CEO"]
  },
  "agency_swarm_version": "1.0.0"
}

Conversation starters appear under data.conversationStarters, and quick-reply phrases appear under data.quickReplies when configured. See Agent Overview.Tool (/tool/<name>):

{
  "response": "Output returned by the tool"
}

Usage tracking

Responses include a usage object with token counts and cost by default. For streaming, the final event: messages payload includes the same usage object.To understand what’s inside usage, see Observability.Usage tracking is configured on the agent (not on FastAPI):

from agency_swarm import Agent, ModelSettings

agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant.",
    model_settings=ModelSettings(include_usage=False),
)

If you’re using LiteLLM models and want usage in streaming responses, keep include_usage=True.

Stream cancellation uses an in-memory registry per process. Use single-worker deployments (e.g., uvicorn workers=1) or sticky routing so cancel requests reach the same worker.

Authentication

Set the environment variable named by app_token_env (default APP_TOKEN) to require Authorization: Bearer <token> on every endpoint. When the variable is absent, authentication is disabled.

Implementation reference

Example: Serving Multiple Agencies and Tools

from agency_swarm import Agency, Agent, function_tool, run_fastapi

# Example tools using agents SDK
@function_tool
def example_tool(example_field: str) -> str:
    """Example tool with input field."""
    return f"Result of ExampleTool operation with {example_field}"

@function_tool
def test_tool(example_field: str) -> str:
    """Test tool with input field."""
    return f"Result of TestTool operation with {example_field}"

# Create agents
agent1 = Agent(name="Assistant1", instructions="You are assistant 1.")
agent2 = Agent(name="Assistant2", instructions="You are assistant 2.")

# Create agency factory functions for proper thread management
def create_agency_1(load_threads_callback=None, save_threads_callback=None):
    return Agency(
        agent1, 
        name="test_agency_1",
        load_threads_callback=load_threads_callback,
        save_threads_callback=save_threads_callback,
    )

def create_agency_2(load_threads_callback=None, save_threads_callback=None):
    return Agency(
        agent2, 
        name="test_agency_2",
        load_threads_callback=load_threads_callback,
        save_threads_callback=save_threads_callback,
    )

run_fastapi(
    agencies={
        "test_agency_1": create_agency_1,
        "test_agency_2": create_agency_2,
    },
    tools=[example_tool, test_tool],
    enable_realtime=True,
    realtime_options={"provider": "openai", "model": "gpt-realtime-2.1-mini"},
)

Endpoints follow the reference above; tool schemas mirror the definitions of example_tool and test_tool.

API Usage Example

You can interact with your agents and tools using HTTP requests:

import requests

agency_url = "http://127.0.0.1:8000/test_agency_1/get_response"
payload = {
    "message": "Hello",
    "client_config": {
        "base_url": "https://my-openai-gateway.example.com/v1",
        "api_key": "sk-...",  # override per request
    },
}

headers = {
    "Authorization": "Bearer 123"  # Replace with your actual token if needed
}

agency_response = requests.post(agency_url, json=payload, headers=headers)
print("Status code:", agency_response.status_code)
print("Response:", agency_response.json())

Request payload

Field	Type	Required	Description
`message`	string or structured list	Yes	User message to start or continue the conversation. Use a structured Responses input list for inline `input_text`, `input_image`, or `input_file` content.
`chat_history`	list	No	Flat list of prior messages with metadata (`agent`, `callerAgent`, `timestamp`) to preserve context.
`recipient_agent`	string	No	Target agent when you want to direct the next turn.
`file_ids`	list	No	IDs of already uploaded files to attach.
`file_urls`	object	No	`{filename: url_or_absolute_path}` map. Local paths require `allowed_local_file_dirs` on `run_fastapi`.
`additional_instructions`	string	No	Extra guidance for the current request.
`user_context`	object	No	Structured data passed to Agency Context without exposing it to the LLM.
`client_config`	object	No	Override `base_url` / `api_key` for this request (and optional `litellm_keys` for `litellm/` models).

How user_context is applied

Merges with any user_context set on the agency instance.
Useful for structured data (ids, preferences, feature flags) you do not want in the prompt.
Accessible within tools for the duration of the run. See Agency Context.

How client_config is applied

Applies only to OpenAI models (no prefix or openai/...) and LiteLLM models (litellm/...).
Custom Model subclasses are not modified; the request still runs, but the override is skipped.
For LiteLLM, you can provide litellm_keys to pass different keys per provider. Requires LiteLLM installed.

client_config fields

base_url (string, optional): Override the API base URL for this request.
api_key (string, optional): Override the API key for this request.
litellm_keys (object, optional): Only for litellm/... models.
- Map provider_name → api_key.
- Example: {"anthropic": "...", "gemini": "..."}

client_config examples

OpenAI model override (gpt-4o):

{
  "message": "Hello",
  "client_config": {
    "base_url": "https://my-openai-gateway.example.com/v1",
    "api_key": "sk-..."
  }
}

LiteLLM mixed providers (requires openai-agents[litellm]):

{
  "message": "Hello",
  "client_config": {
    "base_url": "https://my-litellm-proxy.example.com",
    "litellm_keys": {
      "anthropic": "sk-ant-...",
      "gemini": "AIza..."
    }
  }
}

Cancelling Active Streams

The streaming endpoint supports cancellation via two methods:

1. Automatic Cancellation on Disconnect

When a client disconnects (tab close, refresh, network failure), the stream is automatically cancelled to preserve token costs.

2. Cancel Endpoint

Call the cancel endpoint with the run_id received from the first event of the streaming response. This will allow you to retrieve intermediate results that were generated before the run cancellation. Optionally include cancel_mode (defaults to immediate):

immediate — stop right away and return messages that were fully generated; the in-progress message is discarded.
after_turn — finish the current turn, then stop.

import requests

# Cancel an active stream
cancel_url = "http://127.0.0.1:8000/test_agency/cancel_response_stream"
payload = {"run_id": "your-run-id", "cancel_mode": "after_turn"}  # cancel_mode is optional
response = requests.post(cancel_url, json=payload)
# Returns: {"ok": True, "run_id": "...", "cancelled": True, "cancel_mode": "after_turn", "new_messages": [...]}

OAuth-enabled agencies

Provide a trusted FastAPI dependency that authenticates the request and returns a stable, non-secret user ID:

from fastapi import Depends

from agency_swarm import run_fastapi
from my_app.auth import User, get_current_user

def get_oauth_user_id(user: User = Depends(get_current_user)) -> str:
    return user.id

run_fastapi(
    agencies={"my-agency": create_agency},
    oauth_user_id_dependency=get_oauth_user_id,
)

If any agent uses MCPServerOAuth, run_fastapi automatically:

Adds /auth/callback and /auth/status/{state} routes
Defers OAuth MCP discovery until the model explicitly invokes authenticate_mcp_server(server_name) (so startup never forces auth)
Emits deterministic OAuth stream events on /get_response_stream:
- event: oauth_redirect with {state, server, auth_url}
- event: oauth_status with {state, server, status} where status is pending, authorized, error:<reason>, or timeout
- SSE keepalive comments every 15 seconds while OAuth is pending: : keepalive <timestamp>
Uses the authenticated ID from oauth_user_id_dependency to store tokens per user
Keeps OAuth callback state in an in-process registry. For multi-worker deployments, an external oauth_registry shares callback state only; token files must also use a shared persistent oauth_token_path. Otherwise, use sticky routing or a single worker.
Requires streaming: POST /{agency}/get_response returns 400 when OAuth-enabled MCP servers are present. Use POST /{agency}/get_response_stream (or enable_agui=True) and buffer the stream if you need a single final response.
Applies a 10-minute OAuth wait timeout per request. On timeout, the stream emits oauth_status = timeout and closes cleanly.

oauth_user_id_dependency is required for OAuth-enabled agencies. Do not trust a user ID read directly from a request header or body. FastAPI wiring remains request-scoped and carries the OAuth prompts automatically.

Binding the callback to a user

The provider redirects the browser to /auth/callback as a plain navigation, so it carries no bearer token. By default the callback therefore matches the authorization code to its pending flow by the state parameter alone. state is 32 bytes of URL-safe randomness, and /auth/status/{state} is already restricted to the user who owns the flow. If your deployment authenticates browsers with a cookie or session, set verify_oauth_callback_user=True and make oauth_user_id_dependency resolve that session. The callback then rejects a code whose pending flow belongs to a different user, which closes the remaining session-fixation gap for anyone who learns a victim’s pending state. Leave it off when the dependency only reads a bearer token: the redirect would fail authentication and every OAuth flow would break. For HostedMCPTool, FastAPI cannot infer whether a remote MCP server is public or OAuth-protected. To enable hosted OAuth, wrap only the protected tool with enable_hosted_mcp_tool_oauth(...) and leave tool_config.authorization empty. The tool is withheld until the model invokes authenticate_mcp_server(server_name), so unrelated messages do not start OAuth. run_fastapi(...) creates an in-memory callback-state registry by default. For multiple workers, an external oauth_registry shares callback state only; configure each Agency with an oauth_token_path on shared persistent storage as well. Otherwise, use sticky routing or a single worker. Public hosted MCP servers should stay unwrapped.

For OAuth-protected FastMCP servers, schema discovery (list_tools) is auth-protected. Discovery is deferred until the model invokes authenticate_mcp_server(server_name), so startup does not force OAuth.

Set redirect_uri on MCPServerOAuth to the publicly reachable URL that forwards to the FastAPI /auth/callback route (for example, your ingress hostname plus /auth/callback). You can also set OAUTH_CALLBACK_URL or <SERVER>_REDIRECT_URI as a fallback. If you do not override it, MCPServerOAuth defaults to http://localhost:8000/auth/callback for local development. In production, you must expose and route the callback path back to the issuing pod (sticky routing or shared state), or the OAuth flow will time out.

Serving Standalone Tools

Expose tools as simple HTTP endpoints for external systems, webhooks, or other agents to call directly without agency orchestration.

from agency_swarm import BaseTool, run_fastapi

class Address(BaseTool):
    street: str
    zip_code: int

    def run(self) -> str:
        return f"{self.street} {self.zip_code}"

run_fastapi(tools=[Address], port=8080)

This creates:

POST /tool/Address — execute the tool
GET /openapi.json — full OpenAPI schema
GET /docs — interactive Swagger UI

Generate OpenAPI schema programmatically

Use ToolFactory.get_openapi_schema() to generate the OpenAPI spec programmatically:

from agency_swarm.tools import ToolFactory

schema = ToolFactory.get_openapi_schema(
    [Address],
    url="https://your-server.com",
    title="My Tools API"
)
print(schema)  # Returns a JSON string

File Attachments

Attach files to agency requests using file_ids or file_urls in the payload:

{
  "message": "Summarize this document",
  "file_urls": {"report.pdf": "https://example.com/report.pdf"}
}

For inline Responses attachments that should not be uploaded through the Files API, pass a structured message:

{
  "message": [
    {
      "role": "user",
      "content": [
        {"type": "input_image", "image_url": "data:image/png;base64,...", "detail": "auto"},
        {"type": "input_text", "text": "Describe this image"}
      ]
    }
  ]
}

With manual chat_history, clients can keep structured attachment parts in history so follow-up turns work without reattaching. This may resend inline attachment content or references; it does not reuse server-managed state, previous_response_id, Conversations, or file_id values. The response includes file_ids_map with the uploaded file IDs:

{
  "response": "...",
  "file_ids_map": {"report.pdf": "file-abc123"}
}

When file_urls is used, Agency Swarm also prepends a system message for that turn that records the original source string for each attached file. That system message is included in new_messages, so if your client persists new_messages as chat history, later turns will keep the original attachment source URL or local path in model context. Supported filetypes: .pdf, .jpeg, .jpg, .gif, .png, .c, .cs, .cpp, .csv, .html, .java, .json, .php, .py, .rb, .css, .js, .sh, .ts, .pkl, .tar, .xlsx, .xml, .zip, .doc, .docx, .md, .pptx, .tex, .txt

Local file paths

To support passing local filepaths in the file_urls field, set allowed_local_file_dirs on run_fastapi:

run_fastapi(agencies=..., allowed_local_file_dirs=["/data/uploads"])

Then pass absolute file paths in file_urls:

{"file_urls": {"doc.pdf": "/data/uploads/doc.pdf"}}

Invalid paths return an error field in the response body.

allowed_local_file_dirs uses strict request-time validation for invalid entries: if an entry exists but is not a directory, local file_urls requests fail with an error. Missing directories are skipped and can be created later. The /get_metadata field allowed_local_file_dirs lists only currently usable directory entries.

Welcome

Core Framework

Additional Features

References

Contributing

Migration

FAQ

Installation

Setting Up FastAPI Endpoints

Example: Create an API endpoint for a single agency

Authentication

Implementation reference

API Usage Example

How user_context is applied

How client_config is applied

client_config fields

Cancelling Active Streams

1. Automatic Cancellation on Disconnect

2. Cancel Endpoint

OAuth-enabled agencies

Binding the callback to a user

Serving Standalone Tools

File Attachments

​Installation

​Setting Up FastAPI Endpoints

​Example: Create an API endpoint for a single agency

​Authentication

​Implementation reference

​API Usage Example

​How user_context is applied

​How client_config is applied

​client_config fields

​Cancelling Active Streams

​1. Automatic Cancellation on Disconnect

​2. Cancel Endpoint

​OAuth-enabled agencies

​Binding the callback to a user

​Serving Standalone Tools

​File Attachments

Installation

Setting Up FastAPI Endpoints

Example: Create an API endpoint for a single agency

Authentication

Implementation reference

API Usage Example

How user_context is applied

How client_config is applied

client_config fields

Cancelling Active Streams

1. Automatic Cancellation on Disconnect

2. Cancel Endpoint

OAuth-enabled agencies

Binding the callback to a user

Serving Standalone Tools

File Attachments