agency-swarm-cn-git/examples/fastapi_integration

FastAPI Integration Example

Full Guide: The canonical FastAPI documentation lives in docs/additional-features/fastapi-integration.mdx. This README only summarizes the runnable sample.

This example demonstrates how to properly integrate Agency Swarm with FastAPI, including:

• Serving agencies via HTTP endpoints
• Serving standalone tools via HTTP endpoints
• Handling streaming responses with Server-Sent Events (SSE)
• Properly propagating agent and callerAgent fields in events
• Managing conversation history across requests
• Accessing OpenAPI schemas for tool integration

Files

• server.py - FastAPI server that exposes an agency with two communicating agents
• client.py - Python client showing how to interact with the API endpoints

Setup

1. Install dependencies:

   pip install agency-swarm[fastapi]
   

2. Set your OpenAI API key:

   export OPENAI_API_KEY="your-key-here"
   

3. Optional: Set authentication token:

   export APP_TOKEN="your-secret-token"
   

Running the Example

Start the Server

python server.py

The server will start on http://localhost:8080 with these endpoints:

• POST /my-agency/get_response - Regular response endpoint
• POST /my-agency/get_response_stream - SSE streaming endpoint
• GET /my-agency/get_metadata - Agency structure metadata

Serving Tools

See the “Serving Standalone Tools” section in docs/additional-features/fastapi-integration.mdx for the full walkthrough. In short, calling run_fastapi(tools=[MyTool]) automatically exposes:

• POST /tool/<ToolName> – executes the tool with validation
• GET /openapi.json – OpenAPI 3.1.0 schema for agencies + tools
• GET /docs / GET /redoc – Swagger UI and ReDoc backed by the same schema

All schemas include nested Pydantic models, so you can connect directly to platforms such as Agencii.ai.

Test with Client

python client.py

This will test all endpoints and show how to:

• Make requests with conversation history
• Handle streaming events
• Extract agent metadata from responses

Verify Tool Schemas

Run the helper script to confirm /openapi.json matches ToolFactory.get_openapi_schema():

python print_openapi_schema.py

The script prints the FastAPI /openapi.json response followed by the ToolFactory schema so you can diff them directly (no assertions or extra output).

Important Notes

Field Propagation

The agent and callerAgent fields should appear in:

• All streaming events (added by streaming_utils.py)
• Final new_messages array in both regular and streaming responses
• Tool call events with corresponding call_id for correlation

Conversation Persistence

To maintain conversation context:

1. Start with empty chat_history: []
2. After each response, append new_messages to your chat history
3. Include the updated chat_history in the next request

Authentication

If APP_TOKEN is set, include it in requests:

headers = {"Authorization": f"Bearer {token}"}
response = requests.post(url, json=payload, headers=headers)

Troubleshooting

Events not streaming

1. Check that the client accepts text/event-stream content type
2. Disable buffering in reverse proxies (nginx: proxy_buffering off)
3. Verify SSE format: each event should have data: prefix

Conversation history errors

1. Ensure chat_history is a flat list of message dictionaries
2. Include all required fields: role, content/text, agent, callerAgent
3. Pass load_threads_callback to the agency factory