FastAPI Integration

pip install "agency-swarm[fastapi]"

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()

message: str
# Entire chat history as a flat list of messages with metadata like 'agent', 'callerAgent' and 'timestamp'
chat_history: list[dict[str, Any]] = None
recipient_agent: str = None
file_ids: list[str] = None
# Files to download and attach to the agent. Should be provided in a form of a mapping: {"filename_1":"url", "filename_2":"url", ...}
file_urls: dict[str, str] = None
additional_instructions: str = None

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],
)

import requests

agency_url = "http://127.0.0.1:8000/test_agency_1/get_response"
payload = {
    "message": "Hello",
}

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())

tool_url = "http://127.0.0.1:8000/tool/example_tool"
payload = {
    "example_field": "test",
}

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