The Agency class orchestrates a collection of Agent instances based on entry points and communication flows.
from agency_swarm import Agency, Agentclass 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.
Advanced Note:Handoff from agency_swarm is the Agency Swarm communication-flow class.
SDKHandoff exposes the raw OpenAI Agents SDK handoff type for advanced interoperability.
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, agency_context_override: AgencyContext | 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 agency_context_override: Run-scoped agency context override (for example, isolated thread history) **kwargs: Additional arguments passed to the target agent's get_response Returns: RunResult: The result of the agent execution chain """
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 """
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 """
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 """