> ## Documentation Index
> Fetch the complete documentation index at: https://agency-swarm.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# FastAPI Integration
> Serving your agencies and tools as APIs with FastAPI.
Agency Swarm supports serving your agencies and tools as production-ready HTTP APIs using [FastAPI](https://fastapi.tiangolo.com/). 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:
```bash theme={null}
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
```python theme={null}
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()
```
| 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. |
* 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)
* `/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`.
**Non-streaming (`/get_response`):**
```json theme={null}
{
"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`):**
```json theme={null}
{
"ok": true,
"run_id": "550e8400-e29b-41d4-a716-446655440000",
"cancelled": true,
"cancel_mode": "immediate",
"new_messages": [...]
}
```
**Metadata (`/get_metadata`):**
```json theme={null}
{
"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](/core-framework/agents/overview).
**Tool (`/tool/`):**
```json theme={null}
{
"response": "Output returned by the tool"
}
```
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](/additional-features/observability#usage-payload).
Usage tracking is configured on the agent (not on FastAPI):
```python theme={null}
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 ` on every endpoint. When the variable is absent, authentication is disabled.
### Implementation reference
```python theme={null}
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],
)
```
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:
```python theme={null}
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())
```
| 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](/additional-features/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](/additional-features/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": "..."}`
**OpenAI model override (gpt-4o):**
```json theme={null}
{
"message": "Hello",
"client_config": {
"base_url": "https://my-openai-gateway.example.com/v1",
"api_key": "sk-..."
}
}
```
**LiteLLM mixed providers (requires `openai-agents[litellm]`):**
```json theme={null}
{
"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.
```python theme={null}
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": [...]}
```
***
## Serving Standalone Tools
Expose tools as simple HTTP endpoints for external systems, webhooks, or other agents to call directly without agency orchestration.
```python theme={null}
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
Use `ToolFactory.get_openapi_schema()` to generate the OpenAPI spec programmatically:
```python theme={null}
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:
```json theme={null}
{
"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`:
```json theme={null}
{
"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:
```json theme={null}
{
"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`
To support passing local filepaths in the `file_urls` field, set `allowed_local_file_dirs` on `run_fastapi`:
```python theme={null}
run_fastapi(agencies=..., allowed_local_file_dirs=["/data/uploads"])
```
Then pass absolute file paths in `file_urls`:
```json theme={null}
{"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.