Skip to main content

​
Agency Class

The Agency class orchestrates a collection of Agent instances based on entry points and communication flows. 
from agency_swarm import Agency, Agent

class Agency:
    def __init__(
        self,
        *entry_point_agents: Agent,
        communication_flows: list[CommunicationFlowEntry] | None = None,
        name: str | None = None,
        shared_instructions: str | None = None,
        shared_tools: list[Tool] | None = None,
        shared_tools_folder: str | None = None,
        shared_files_folder: str | None = None,
        shared_mcp_servers: list[Any] | None = None,
        send_message_tool_class: type | None = None,
        load_threads_callback: ThreadLoadCallback | None = None,
        save_threads_callback: ThreadSaveCallback | None = None,
        user_context: dict[str, Any] | None = None,
    ):
        """
        Initialize an Agency instance.

        Parameters:
            *entry_point_agents: Agent instances serving as entry points for external interaction
            communication_flows: Flexible communication flow definitions supporting tuples and AgentFlow objects
            name: Optional name for the agency
            shared_instructions: Instruction text or file path applied to every agent
            shared_tools: Tool instances or BaseTool classes to add to all agents (each agent gets independent copies)
            shared_tools_folder: Path to folder containing tool files to load and share with all agents
            shared_files_folder: Path to folder of files to upload and share with all agents via a common vector store
            shared_mcp_servers: MCP server instances to attach to all agents
            send_message_tool_class: Fallback SendMessage tool class when a communication flow does not define its own
            load_threads_callback: Callable used to load persisted conversation threads
            save_threads_callback: Callable used to save conversation threads
            user_context: Initial shared context accessible to all agents
        """
Note: CommunicationFlowEntry is a type alias defined in agency_swarm.agency.core that accepts (Agent, Agent) tuples, AgentFlow chains, or those accompanied by a custom SendMessage tool class.

​
Key Attributes

  • agents (dict[str, Agent]): Dictionary mapping agent names to their instances
  • entry_points (list[Agent]): Agents designated as entry points for external interaction
  • thread_manager (ThreadManager): Manager responsible for handling conversation threads
  • event_stream_merger (EventStreamMerger): Aggregates streaming events from concurrent agent runs
  • persistence_hooks (PersistenceHooks | None): Optional hooks for loading/saving thread state
  • shared_instructions (str | None): Instructions prepended to every agent’s system prompt
  • shared_tools (list[Tool] | None): Tool instances or BaseTool classes shared with all agents
  • shared_tools_folder (str | None): Path to folder containing shared tool files
  • shared_files_folder (str | None): Path to folder of files shared via a common vector store
  • shared_mcp_servers (list[Any] | None): MCP server instances attached to all agents
  • user_context (dict[str, Any]): Shared user-defined context accessible within MasterContext
  • send_message_tool_class (type | None): Fallback SendMessage tool when no flow-specific tool is set

​
Key Methods

async def get_response(
    self,
    message: str | list[TResponseInputItem],
    recipient_agent: str | Agent | None = None,
    context_override: dict[str, Any] | None = None,
    hooks_override: RunHooks | None = None,
    run_config: RunConfig | None = None,
    file_ids: list[str] | None = None,
    additional_instructions: str | None = None,
    **kwargs: Any,
) -> RunResult:
    """
    Initiate an interaction with a specified agent within the agency.

    Parameters:
        message: The input message for the agent
        recipient_agent: Target agent instance or name (defaults to first entry point)
        context_override: Additional context to pass to the agent run
        hooks_override: Specific hooks to use for this run, overriding agency defaults
        run_config: Configuration for the agent run
        file_ids: Additional file IDs for the agent run
        additional_instructions: Additional instructions for this run only
        **kwargs: Additional arguments passed to the target agent's get_response

    Returns:
        RunResult: The result of the agent execution chain
    """

​
Agent Class

The Agent class extends the base agents.Agent with capabilities for multi-agent collaboration within an Agency. 
from agency_swarm import Agent, ModelSettings

class Agent(BaseAgent[MasterContext]):
    def __init__(self, **kwargs: Any):
        """
        Initialize an Agency Swarm Agent.

        Parameters:
            name (str): Agent name (**required**)
            instructions (str | Path | None): System prompt text or path
            description (str | None): Optional role description for send_message tools
            files_folder (str | Path | None): Folder for local file management and vector stores
            tools_folder (str | Path | None): Directory for automatic tool discovery
            schemas_folder (str | Path | None): Directory containing OpenAPI schemas for tool generation
            api_headers (dict[str, dict[str, str]] | None): Per-schema HTTP headers for generated tools
            api_params (dict[str, dict[str, Any]] | None): Per-schema default parameters for generated tools
            include_search_results (bool): Include search results in FileSearchTool output (default False)
            validation_attempts (int): Number of retries when output guardrails trigger (default 1)
            throw_input_guardrail_error (bool): Controls input guardrail behavior—False enables friendly mode (guidance as assistant message), True enables strict mode (raises exceptions). Default: False
            handoff_reminder (str | None): Custom reminder text when handing off to another agent
            model (str | Model | None): Model identifier; defaults to agents SDK configuration
            model_settings (ModelSettings | None): Model configuration overrides
            tools (list[Tool] | None): Explicit tool instances to register
            mcp_servers (list[MCPServer] | None): Model Context Protocol servers to attach
            mcp_config (MCPConfig | None): Configuration for MCP servers
            input_guardrails (list[InputGuardrail] | None): Pre-execution validation hooks
            output_guardrails (list[OutputGuardrail] | None): Post-execution validation hooks
            output_type (type[Any] | AgentOutputSchemaBase | None): Type of final output structure
            hooks (AgentHooks | None): Lifecycle callbacks from the agents SDK
            tool_use_behavior ("run_llm_again" | "stop_on_first_tool" | StopAtTools | dict[str, Any] | Callable):
                Tool execution policy passed through to the agents SDK
            reset_tool_choice (bool | None): Whether to reset tool choice after tool calls
        """

​
Key Attributes

  • files_folder (str | Path | None): Local folder for file management and vector stores
  • tools_folder (str | Path | None): Directory for automatic tool discovery and loading
  • schemas_folder (str | Path | None): Folder scanned for OpenAPI schemas to auto-generate tools
  • api_headers / api_params (dict): Per-schema metadata applied when generating tools
  • description (str | None): Agent role description for dynamic send_message tools
  • include_search_results (bool): Whether FileSearchTool responses include citation context
  • validation_attempts (int): Retry count for output guardrail enforcement
  • throw_input_guardrail_error (bool): Controls input guardrail mode—False for friendly (guidance as assistant), True for strict (raises exceptions)
  • handoff_reminder (str | None): Custom reminder appended to handoff prompts
  • tool_concurrency_manager (ToolConcurrencyManager): Coordinates concurrent tool execution

​
Core Execution Methods

async def get_response(
    self,
    message: str | list[TResponseInputItem],
    sender_name: str | None = None,
    context_override: dict[str, Any] | None = None,
    hooks_override: RunHooks | None = None,
    run_config_override: RunConfig | None = None,
    file_ids: list[str] | None = None,
    additional_instructions: str | None = None,
    agency_context: AgencyContext | None = None,
    **kwargs: Any,
) -> RunResult:
    """
    Run the agent's turn in the conversation loop.

    Parameters:
        message: Input message as string or structured input items list
        sender_name: Name of sending agent (None for user interactions)
        context_override: Optional context data to override default MasterContext values
        hooks_override: Optional hooks to override default agent hooks
        run_config_override: Optional run configuration settings
        file_ids: List of OpenAI file IDs to attach to the message
        additional_instructions: Additional instructions for this run only
        agency_context: Injected AgencyContext (auto-created when running standalone)
        **kwargs: Additional keyword arguments forwarded to the agents SDK (e.g., max_turns)

    Returns:
        RunResult: The complete execution result from the agents SDK
    """

​
Tool Management

def add_tool(self, tool: Tool) -> None:
    """
    Add a Tool instance to the agent's list of tools.

    Parameters:
        tool: The agents.Tool instance to add

    Raises:
        TypeError: If tool is not an instance of agents.Tool
    """

​
File Management

def upload_file(self, file_path: str, include_in_vector_store: bool = True) -> str:
    """
    Upload a file using the agent's file manager.

    Parameters:
        file_path: Path to the file to upload
        include_in_vector_store: Whether to add file to vector store

    Returns:
        str: File ID of the uploaded file

    Raises:
        RuntimeError: If the agent has no file manager configured
    """