Agent Module

The agent module provides the core Agent class for creating AI agents.

Agent Class

class pantheon.agent.Agent(name: str, instructions: str, model: str | list[str] | None = None, model_params: dict | None = None, icon: str = '🤖', tools: list[Callable] | None = None, response_format: Any | None = None, use_memory: bool = True, memory: Memory | None = None, tool_timeout: int | None = None, relaxed_schema: bool = False, max_tool_content_length: int | None = None, description: str | None = None)

Bases: object

The Agent class is the core component of Pantheon, providing a flexible interface for creating AI-powered agents with tools, memory, and collaboration capabilities.

Parameters:

• name – The name of the agent.

• instructions – The instructions for the agent. The instructions are the system instructions that the agent will follow. All prompt composition (work strategy, delegation, skills, etc.) should be done at template parsing time, not at runtime.

• model – The model to use for the agent. Can be a single model or list of fallback models.

• model_params – The additional parameters for the model(LLM).

• icon – The icon to use for the agent.

• tools – The tools to use for the agent.

• response_format – The response format to use for the agent. It can be a Pydantic model or a function that returns a Pydantic model.

• use_memory – Whether to use memory for the agent. (default: True)

• memory – The memory to use for the agent. If not provided, a new memory will be created.

• tool_timeout – The timeout for the tool. (default: from settings.endpoint.local_toolset_timeout, or 3600s)

• relaxed_schema – Use relaxed (non-strict) tool schema mode. (default: False)

• max_tool_content_length – The maximum length of the tool content. (default: 100000)

• description – The description of the agent. (default: None)

__init__(name: str, instructions: str, model: str | list[str] | None = None, model_params: dict | None = None, icon: str = '🤖', tools: list[Callable] | None = None, response_format: Any | None = None, use_memory: bool = True, memory: Memory | None = None, tool_timeout: int | None = None, relaxed_schema: bool = False, max_tool_content_length: int | None = None, description: str | None = None)

property tool_timeout: int

Public API property for backward compatibility and dynamic access.

property max_tool_content_length: int

Public API property for backward compatibility and dynamic access.

property functions: dict[str, Callable]

Get current tools (returns a copy for safety).

tool(func: Callable, key: str | None = None)

Add a tool to the agent.

Parameters:

• func – The tool function to add to the agent.

• key – The key to use for the tool. If not provided, the name of the function will be used.

Returns:

The agent instance.

remove_tool(key: str) → bool

Remove a tool from the agent.

Parameters:

key – The name of the tool to remove.

Returns:

True if the tool was found and removed, False otherwise.

Return type:

bool

async toolset(toolset: ToolSet | ToolProvider) → Agent

Add a toolset to the agent (supports both ToolSet and ToolProvider)

Behavior: - ToolSet: Automatically wrapped in LocalProvider for unified provider-based routing - ToolProvider (LocalProvider/ToolSetProvider): Dynamic routing - tools retrieved on-demand

Parameters:

toolset – Either a ToolSet instance (will be wrapped in LocalProvider) or a ToolProvider instance (used directly)

Returns:

The agent instance

async mcp(name: str, provider: ToolProvider) → Agent

Add a MCP provider to the agent (one at a time)

Dynamic routing approach: Provider tools are retrieved on-demand via get_tools_for_llm() and called via call_tool() with prefix routing. No pre-wrapping.

Parameters:

• name – Name/identifier for this provider (e.g., ‘biomcp’)

• provider – ToolProvider instance (MCPProvider recommended, already initialized)

Returns:

The agent instance

async get_tools_for_llm() → list[dict]

Get all tools for LLM - includes _base_functions and provider tools

Dynamically retrieves tools from providers (utilizing their caching mechanisms). Provider tools are prefixed with {provider_name}_ to distinguish them from agent’s own tools.

Returns:

List of tool definitions in OpenAI format

async call_tool(prefixed_name: str, args: dict, context_variables: dict | None = None, tool_call_id: str | None = None) → Any

Call a tool by name with automatic routing and suffix matching fallback.

async run(msg: str | BaseModel | AgentResponse | list[str | BaseModel | dict] | AgentTransfer | VisionInput, response_format: Any | None = None, tool_use: bool = True, context_variables: dict | None = None, process_chunk: Callable | None = None, process_step_message: Callable | None = None, check_stop: Callable | None = None, memory: Memory | None = None, use_memory: bool | None = None, update_memory: bool = True, tool_timeout: int | None = None, model: str | list[str] | None = None, allow_transfer: bool = True, execution_context_id: str | None = None) → AgentResponse | AgentTransfer

Run the agent.

Parameters:

• msg – The input message to the agent.

• response_format – The response format to use.

• tool_use – Whether to use tools.

• context_variables – Runtime variables available to tools during execution. These are stored in ExecutionContext.runtime_context and injected into tools that request them via the ‘context_variables’ parameter. Example: {‘user_id’: ‘123’, ‘session_id’: ‘abc’}

• process_chunk – The function to process the chunk.

• process_step_message – The function to process the step message.

• check_stop – The function to check if the agent should stop.

• memory – The memory to use.

• use_memory – Whether to use short term memory.

• update_memory – Whether to update the short term memory.

• tool_timeout – The timeout for the tool.

• model – The model to use in this run. Could be a list of models for fallback. If not provided, the model will be selected from the agent’s models.

• allow_transfer – Whether to allow transfer to another agent.

• execution_context_id – Unique identifier for sub-agent delegation contexts. Used for message filtering in _run_stream to isolate messages by delegation context. None for primary/team agents.

Returns:

The agent response. Either an AgentResponse or an AgentTransfer. If the agent is interrupted, the AgentResponse will have the interrupt flag set to True. If the agent is transferring to another agent, the AgentTransfer will be returned.

async run_loop(process_chunk: Callable | None = None, process_step_message: Callable | None = None, check_stop: Callable | None = None, context_variables: dict | None = None, memory: Memory | None = None, tool_timeout: int | None = None, model: str | list[str] | None = None, on_response: Callable | None = None, on_error: Callable | None = None) → None

Persistent event-driven loop consuming messages from input_queue.

Blocks until stop_loop() is called or the task is cancelled. Each message dequeued triggers a full agent.run() cycle. Background task completions auto-enqueue notifications, so the agent is automatically triggered when bg tasks finish.

Parameters:

• process_chunk – Streaming chunk callback (forwarded to run()).

• process_step_message – Step message callback (forwarded to run()).

• check_stop – Stop check callback (forwarded to run()).

• context_variables – Shared context variables for all runs.

• memory – Memory instance to use.

• tool_timeout – Tool timeout (forwarded to run()).

• model – Model override (forwarded to run()).

• on_response – Callback(AgentResponse) after each successful run.

• on_error – Callback(Exception) when a run fails.

stop_loop()

Signal run_loop() to exit after current iteration.

setup_bg_notify_queue(queue: Queue)

Wire bg task completion to an external asyncio.Queue.

Alternative to run_loop() for frontends that manage their own event loops (e.g. REPL with prompt_toolkit, ChatRoom server).

Parameters:

queue – The asyncio.Queue to push notification strings into.

async chat(message: str | dict | None = None)

Chat with the agent with a REPL interface.

async serve(**kwargs)

Serve the agent to a remote server.

Constructor Parameters

Parameter	Type	Description
name	str	Unique name for the agent
instructions	str	System instructions defining agent behavior
model	str | list[str]	LLM model(s) to use (default: “gpt-4.1-mini”)
icon	str	Display icon for the agent (default: ‘🤖’)
tools	list[Callable]	List of tool functions the agent can use
response_format	Any	Expected response format (for structured outputs)
use_memory	bool	Enable memory persistence (default: True)
memory	Memory	Custom memory instance
tool_timeout	int	Tool execution timeout in seconds (default: 600)
relaxed_schema	bool	Use relaxed (non-strict) tool schema mode (default: False)
max_tool_content_length	int | None	Maximum length for tool outputs (default: 100000)

Methods

The Agent class provides the following key methods:

• tool(func): Decorator to add a tool function to the agent

• chat(): Start an interactive chat session

• run(messages): Run the agent with given messages

• remote_toolset(service_id): Connect to a remote toolset

Response Classes

class pantheon.agent.AgentResponse(*, agent_name: str, content: Any, details: ResponseDetails | None, interrupt: bool = False)

Bases: BaseModel

The AgentResponse class is used to store the agent response.

agent_name: str

content: Any

details: ResponseDetails | None

interrupt: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pantheon.agent.AgentTransfer(*, from_agent: str, to_agent: str, history: list[dict], context_variables: dict, init_message_length: int, tool_call_id: str | None = None)

Bases: BaseModel

The AgentTransfer class is used to transfer the agent response to another agent.

from_agent: str

to_agent: str

history: list[dict]

context_variables: dict

init_message_length: int

tool_call_id: str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Type Definitions

pantheon.agent.AgentInput

Type alias for valid agent inputs:

• str: Simple text message

• BaseModel: Structured input

• AgentResponse: Response from another agent

• list[str | BaseModel | dict]: Multiple inputs

• AgentTransfer: Transfer from another agent

• VisionInput: Image input

alias of str | BaseModel | AgentResponse | list[str | BaseModel | dict] | AgentTransfer | VisionInput

Examples

Basic Agent

from pantheon.agent import Agent

agent = Agent(
    name="assistant",
    instructions="You are a helpful AI assistant."
)

# Interactive chat
await agent.chat()

Agent with Tools

from pantheon.agent import Agent

agent = Agent(
    name="calculator",
    instructions="You help with calculations."
)

@agent.tool
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

@agent.tool
def multiply(a: float, b: float) -> float:
    """Multiply two numbers."""
    return a * b

# Run with tools
result = await agent.run("What is 5 + 3?")
print(result.content)

Agent with Memory

from pantheon.agent import Agent
from pantheon.memory import Memory

# Custom memory
memory = Memory(agent_id="unique_id")

agent = Agent(
    name="memory_bot",
    instructions="Remember our conversations.",
    memory=memory
)

# First interaction
await agent.run("My name is Alice")

# Later interaction (remembers context)
result = await agent.run("What's my name?")
print(result.content)  # Should mention "Alice"

Multi-Model Fallback

agent = Agent(
    name="robust_agent",
    instructions="Handle tasks reliably."
)

Remote Toolset

from pantheon.toolsets.python import PythonInterpreterToolSet
from pantheon.toolsets.utils.toolset import run_toolsets

async def setup():
    toolset = PythonInterpreterToolSet("python")

    async with run_toolsets([toolset]):
        agent = Agent(
            name="coder",
            instructions="You can run Python code."
        )
        await agent.remote_toolset(toolset.service_id)

        result = await agent.run("Calculate fibonacci(10)")
        print(result.content)