Agent API#

The Agent class is the fundamental building block of Pantheon.

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)[source][source]#

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)[source][source]#
events_queue: Queue[source]#
input_queue: Queue[source]#
providers: dict[str, ToolProvider][source]#
not_loaded_toolsets: list[str][source]#
context_injectors: list[source]#
property tool_timeout: int[source]#

Public API property for backward compatibility and dynamic access.

property max_tool_content_length: int[source]#

Public API property for backward compatibility and dynamic access.

property functions: dict[str, Callable][source]#

Get current tools (returns a copy for safety).

tool(func: Callable, key: str | None = None)[source][source]#

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[source][source]#

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[source][source]#

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[source][source]#

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][source][source]#

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[source][source]#

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[source][source]#

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[source][source]#

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()[source][source]#

Signal run_loop() to exit after current iteration.

setup_bg_notify_queue(queue: Queue)[source][source]#

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)[source][source]#

Chat with the agent with a REPL interface.

async serve(**kwargs)[source][source]#

Serve the agent to a remote server.

Creating an Agent#

Basic Agent

from pantheon.agent import Agent

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

With Toolsets

from pantheon.toolsets import FileManagerToolSet, ShellToolSet

agent = Agent(
    name="developer",
    instructions="You are a developer."
)

# Add toolsets using await agent.toolset()
await agent.toolset(FileManagerToolSet("files"))
await agent.toolset(ShellToolSet("shell"))

With Memory

agent = Agent(
    name="assistant",
    instructions="...",
    use_memory=True
)

Constructor Parameters#

Parameter

Type

Description

name

str

Agent name (used for identification)

instructions

str

System prompt defining agent behavior

model

str | list[str]

Model specification: quality tag ("high", "normal", "low"), capability combo ("high,vision"), specific model ("openai/gpt-4o"), or fallback list

tools

list[Callable]

List of callable functions (for ToolSets, use await agent.toolset())

use_memory

bool

Enable conversation persistence

max_iterations

int

Maximum tool call iterations (default: 10)

Methods#

run()#

Execute a single query and return the response.

response = await agent.run("What is 2 + 2?")
print(response.content)

Parameters:

  • message (str): The user message

Returns: AgentResponse

chat()#

Start an interactive REPL session.

await agent.chat()

stream()#

Stream the response token by token.

async for chunk in agent.stream("Tell me a story"):
    print(chunk.content, end="")

toolset()#

Add a toolset to the agent at runtime.

from pantheon.toolsets import FileManagerToolSet, ShellToolSet

agent = Agent(
    name="developer",
    instructions="You are a developer."
)

# Add toolsets dynamically
await agent.toolset(FileManagerToolSet("files"))
await agent.toolset(ShellToolSet("shell"))

Parameters:

  • toolset (ToolSet | ToolProvider): The toolset or provider to add

Returns: Agent (for method chaining)

mcp()#

Add an MCP (Model Context Protocol) provider to the agent.

from pantheon.providers import MCPProvider

# Add MCP provider
await agent.mcp(
    name="custom_tools",
    provider=MCPProvider(uri="stdio://path/to/mcp/server")
)

Parameters:

  • name (str): Name for the MCP provider

  • provider (ToolProvider): The MCP provider instance

Returns: Agent (for method chaining)

AgentResponse#

The response object returned by run():

response = await agent.run("...")

# Content
print(response.content)

# Messages (full conversation)
print(response.messages)

# Tool calls made
print(response.tool_calls)

# Token usage
print(response.usage)

Custom Tools#

Using Decorator

@agent.tool
def my_function(param: str) -> str:
    """Description for the LLM."""
    return f"Result: {param}"

Using ToolSet

from pantheon.toolset import ToolSet, tool

class MyTools(ToolSet):
    def __init__(self, name: str):
        super().__init__(name)

    @tool
    def calculate(self, expression: str) -> float:
        """Evaluate math expression."""
        return eval(expression)

# Add toolset at runtime
agent = Agent(name="calculator", instructions="...")
await agent.toolset(MyTools("math"))

Model Selection#

Pantheon supports smart model selection with quality tags, or you can specify exact models.

Smart Selection (Recommended)

Use quality tags to let Pantheon choose the best available model:

# Quality tags
agent = Agent(model="high")      # Best quality model
agent = Agent(model="normal")    # Balanced (default)
agent = Agent(model="low")       # Fast and cheap

# With capability requirements
agent = Agent(model="high,vision")      # High quality + vision
agent = Agent(model="normal,reasoning") # Normal + reasoning
agent = Agent(model="low,tools")        # Cheap + function calling

Specific Models

Use provider/model format for exact model selection:

# OpenAI
agent = Agent(model="openai/gpt-5.4")
agent = Agent(model="openai/gpt-5-mini")

# Anthropic
agent = Agent(model="anthropic/claude-opus-4-5-20251101")
agent = Agent(model="anthropic/claude-sonnet-4-5-20250929")

# Other providers (via native SDK adapters)
agent = Agent(model="gemini/gemini-3-pro-preview")
agent = Agent(model="deepseek/deepseek-chat")
agent = Agent(model="mistral/mistral-large")

Fallback Chains

Provide multiple models for automatic failover:

agent = Agent(model=["openai/gpt-5.4", "openai/gpt-5.2", "openai/gpt-4o"])

See Model Configuration for full model configuration details.

Best Practices#

Clear Instructions

Write specific, clear instructions:

agent = Agent(
    instructions="""You are a Python developer assistant.

    Your responsibilities:
    - Write clean, well-documented code
    - Follow PEP 8 style guidelines
    - Include error handling

    When asked to write code:
    1. First explain your approach
    2. Write the code
    3. Explain how to use it"""
)

Appropriate Model

  • Use quality tags for portability: "high", "normal", "low"

  • Use capability tags when needed: "high,vision", "normal,reasoning"

  • Use specific models only when you need exact behavior

Tool Selection

Only include tools the agent needs:

# Good - specific tools
agent = Agent(name="dev", instructions="...")
await agent.toolset(FileManagerToolSet("files"))

# Avoid - too many tools can confuse the model
await agent.toolset(ToolSet1("t1"))
await agent.toolset(ToolSet2("t2"))
await agent.toolset(ToolSet3("t3"))
# ...