Agentor class API reference

Overview

The Agentor class is the main entry point for building AI agents. It wires a model, some tools, and the loop that drives them, then lets you run the agent, stream it, resume it, or serve it as an API.

Constructor

Agentor(
    name: str,
    instructions: Optional[str] = None,
    model: Any = "gpt-5-nano",
    tools: Optional[List[Any]] = None,
    output_type: Any = None,
    debug: bool = False,
    api_key: Optional[str] = None,
    model_settings: Optional[ModelSettings] = None,
    skills: Optional[List[str]] = None,
    enable_tracing: bool = False,
    max_turns: int = 20,
    max_tool_failures: int = 2,
    store: Any = None,
    base_url: Optional[str] = None,
    tracer: Any = None,
    trace_group_id: Optional[str] = None,
    trace_metadata: Optional[Dict[str, Any]] = None,
    engine: Optional[Literal["native"]] = None,
)

Parameters

str

required

The name of the agent. Used in API endpoints, A2A agent cards, and traces.

str

default:"None"

System prompt that defines the agent’s behavior. Sent as the first message of every run.

str | Model

default:"gpt-5-nano"

The model to use. A bare name such as "gpt-4o-mini" goes to OpenAI. A provider/model string such as "gemini/gemini-2.5-flash" is routed through LiteLLM. With base_url set, the string is passed to that provider untouched. See Model providers.

List[Callable | Tool | str | BaseTool | MCPServer]

default:"None"

What the agent can call. Any mix of:

Functions decorated with @function_tool, or plain callables with type hints
String names from the tool registry — currently "get_weather" and "current_datetime"
BaseTool subclasses, whose @capability methods each become a tool
MCPServer instances, connected for the duration of each run

Provider-hosted tools (web search, file search, hosted MCP) are not supported. They have no callable body Agentor can invoke, so passing one raises TypeError. Write a function tool instead.

type[BaseModel]

default:"None"

A Pydantic model describing the shape you want back. result.final_output becomes an instance of it. See Structured output.

bool

default:"False"

Accepted for backwards compatibility. It currently has no effect — use Python’s logging to see engine activity.

str

default:"None"

API key for the provider. Falls back to the OPENAI_API_KEY environment variable.

ModelSettings

default:"None"

Generation parameters: temperature, max_tokens, top_p, and more. See ModelSettings.

List[str]

default:"None"

Paths to skill files, injected into the system prompt. See Skills.

bool

default:"False"

Turn on Celesto tracing for this agent. Requires CELESTO_API_KEY, and raises if it is missing. Tracing is off by default; setting CELESTO_API_KEY alone does not enable it. See Tracing.

int

default:"20"

How many model calls a run may make before giving up. A run that hits the limit ends with status="max_turns" instead of raising.

int

default:"2"

How many times any one tool may raise before Agentor stops offering it to the model. Applies per tool, per run — so a run with two tools has a budget of max_tool_failures for each. Set higher for tools that are legitimately flaky, or lower to fail fast.

Store

default:"None"

Where to save the run’s events, so an interrupted run can be resumed. Use FileStore or MemoryStore from agentor.engine.store. See Durable runs.

str

default:"None"

An OpenAI-compatible endpoint to talk to instead of OpenAI — OpenRouter, Groq, Together, vLLM, Ollama, and others. See Model providers.

CelestoTracer

default:"None"

A tracer built with setup_celesto_tracing. Passing one turns tracing on for the agent — you do not also need enable_tracing=True. See Tracing.

str

default:"None"

Optional group id attached to every trace this agent exports. Traces sharing an id are grouped together in the Celesto dashboard, so use it to correlate the runs that make up one user session, one job, or one request. See Tracing.

Dict[str, Any]

default:"None"

Optional dictionary of key-value pairs attached to every trace this agent exports. Useful for tagging environment, tenant, or experiment so the dashboard can filter by it.

Literal["native"]

default:"None"

Vestigial. "native" is accepted and does nothing; any other value raises. There is one engine and it is the default.

engine="agents" raised in 0.1.0. The openai-agents engine was removed — drop the argument.

Methods

run

Run the agent synchronously.

def run(input: str, tracing: Optional[bool] = None) -> RunResult

Parameters:

input (str): the prompt
tracing (bool | None): override tracing for this call. None keeps the agent’s configuration, False sends nothing for this run, True traces it even when the agent has tracing off (requires CELESTO_API_KEY). See Tracing.

Returns: a RunResult.

from agentor import Agentor

agent = Agentor(
    name="Assistant",
    instructions="You are a helpful assistant",
)

result = agent.run("Write a haiku about recursion in programming.")
print(result.final_output)

run() cannot be called from inside a running event loop. Use await agent.arun(...) there.

arun

Run the agent asynchronously, with batching and model fallbacks.

async def arun(
    input: list[str] | str | list[AgentInputType],
    limit_concurrency: int = 10,
    max_turns: Optional[int] = None,
    fallback_models: Optional[List[str]] = None,
    tracing: Optional[bool] = None,
) -> RunResult | List[RunResult]

Parameters:

input: one prompt, a list of prompts to run concurrently, or a list of message dicts to continue a conversation
limit_concurrency (int): maximum concurrent runs when input is a list of prompts (default: 10)
max_turns (int): turn budget for this call. Defaults to the agent’s max_turns
fallback_models (List[str]): models to try if the primary one is rate limited or errors. Configured temperature and token limits carry across
tracing (bool | None): override tracing for this call. None keeps the agent’s configuration, False sends nothing, True traces it even when the agent has tracing off. When input is a batch, the flag applies to every prompt

import asyncio
from agentor import Agentor

agent = Agentor(name="Assistant", model="gpt-5-mini")

# Single prompt
result = await agent.arun("What is the weather in London?")
print(result.final_output)

# Batch — each prompt is its own run, executed concurrently
results = await agent.arun([
    "What is the weather in London?",
    "What is the weather in Paris?",
])
for result in results:
    print(result.final_output)

# With fallback models
result = await agent.arun(
    "Analyze this dataset",
    fallback_models=["gpt-4o-mini", "gpt-4o"],
)

A batch call uses asyncio.gather(..., return_exceptions=True), so a failed prompt yields an exception object in the list rather than raising. Check each item’s type before using it.

resume

Continue a run that was saved by a store.

def resume(run_id: str) -> RunResult
async def aresume(run_id: str) -> RunResult

Parameters:

run_id (str): the id from result.run_id

Returns: a RunResult covering the whole run, including the part that ran before the interruption.

from agentor import Agentor
from agentor.engine.store import FileStore

agent = Agentor(name="Agent", tools=[...], store=FileStore("runs"))

result = agent.run("Do the thing")
agent.resume(result.run_id)   # after a crash; a finished run is returned as-is

Raises ValueError if the agent has no store, and KeyError if the store has no run with that id.

fork

Copy a persisted run to a new id and continue it there — even a run that already finished, which resume() refuses to re-execute. See Durable runs.

def fork(
    run_id: str,
    input: Optional[str | list[dict]] = None,
    *,
    fork_id: Optional[str] = None,
    max_turns: Optional[int] = None,
    tracing: Optional[bool] = None,
) -> RunResult

async def afork(
    run_id: str,
    input: Optional[str | list[dict]] = None,
    *,
    fork_id: Optional[str] = None,
    max_turns: Optional[int] = None,
    tracing: Optional[bool] = None,
) -> RunResult

Parameters:

run_id (str): the run to branch from
input (str | list[dict] | None): a prompt or message list to continue the fork with right away. None just copies the run under a new id — an unfinished parent is completed, a finished one comes back as a snapshot with no model call
fork_id (str): choose the new run’s id yourself. Defaults to a generated one
max_turns (int): turn budget for this call, same as run(). Applies to the fork’s continuation only
tracing (bool | None): override tracing for this call, same semantics as run(). Applies to the fork’s continuation only

Returns: a RunResult for the new run — result.run_id is never the parent’s.

from agentor import Agentor
from agentor.engine.store import FileStore

agent = Agentor(name="Agent", tools=[...], store=FileStore("runs"))

result = agent.run("Draft a launch plan")
variant = agent.fork(result.run_id, "Make it punchier")  # new run_id; result is untouched

Raises ValueError if the agent has no store, if fork_id is already in use, or if input is an empty list. Raises KeyError if the store has no run with that run_id.

The fork keeps the parent’s entire trace — every generation with its reasoning, when the model returned any, every tool call and result — and then lives its own life. Nothing appended to the fork touches the parent, and vice versa.

chat

Run the agent, optionally returning a stream.

async def chat(
    input: str,
    stream: bool = False,
    serialize: bool = True,
    tracing: Optional[bool] = None,
)

Parameters:

input (str): user message
stream (bool): return an async iterator instead of a RunResult
serialize (bool): when streaming, yield JSON strings (default) or AgentOutput objects
tracing (bool | None): override tracing for this call, same semantics as run

Returns: a RunResult, or an async iterator when stream=True.

stream_chat

Stream the agent’s progress as it works.

async def stream_chat(
    input: str,
    serialize: bool = True,
    tracing: Optional[bool] = None,
) -> AsyncIterator[Union[str, AgentOutput]]

Parameters:

input (str): user message
serialize (bool): yield JSON strings (default) or AgentOutput objects
tracing (bool | None): override tracing for this call, same semantics as run

import asyncio
from agentor import Agentor

agent = Agentor(name="Assistant", model="gpt-5-mini")

async def main():
    async for event in agent.stream_chat("Tell me a story"):
        print(event, flush=True)

asyncio.run(main())

See the streaming guide for the event shape.

serve

Serve the agent as an HTTP API with A2A protocol support.

def serve(
    host: Literal["0.0.0.0", "127.0.0.1", "localhost"] = "0.0.0.0",
    port: int = 8000,
    log_level: Literal["debug", "info", "warning", "error"] = "info",
    access_log: bool = True,
)

from agentor import Agentor

agent = Agentor(
    name="Weather Agent",
    model="gpt-5-mini",
    instructions="You are a helpful weather assistant.",
)

# Serves at http://0.0.0.0:8000
# Agent card at http://0.0.0.0:8000/.well-known/agent-card.json
agent.serve(port=8000)

from_md

Create an agent from a markdown file with YAML frontmatter.

@classmethod
def from_md(
    cls,
    md_path: str | Path,
    *,
    model: Any = None,
    tools: Optional[List[Any]] = None,
    output_type: Any = None,
    debug: bool = False,
    api_key: Optional[str] = None,
    model_settings: Optional[ModelSettings] = None,
) -> Agentor

Keyword arguments override the frontmatter.

weather_agent.md

---
name: Weather Agent
tools: ["get_weather", "current_datetime"]
model: gpt-4o-mini
temperature: 0.3
---

You are a helpful weather assistant with access to real-time data.

from agentor import Agentor

agent = Agentor.from_md("agents/weather_agent.md")
result = agent.run("What's the weather in Tokyo?")

Tool names in the frontmatter must exist in the registry. Unknown names are skipped with a warning, so an agent can end up with fewer tools than the file lists.

think

Run the agent through a chain-of-thought prompt.

def think(query: str) -> str

Returns: the agent’s reasoning and conclusion as text.

RunResult

Every run returns a RunResult:

str | BaseModel | None

The answer. A parsed model instance when output_type is set, None if the run did not finish.

'completed' | 'max_turns' | 'failed'

How the run ended.

str | None

Why it did not complete. None on success.

str | None

Set when the agent has a store. Pass it to resume() or fork().

Usage

input_tokens, output_tokens, and total_tokens for the run.

list[dict]

The conversation as the model saw it. Feed it back into arun() to continue.

list[Event]

Every step of the run, in order.

list[Event]

Just the tool calls, with the arguments the model chose.

str(result) returns final_output, so print(result) prints the answer.

Usage Examples

Basic agent

from agentor import Agentor

agent = Agentor(
    name="Assistant",
    instructions="You are a helpful assistant",
)

result = agent.run("Explain quantum computing in simple terms")
print(result.final_output)

Agent with a different provider

import os
from agentor import Agentor

agent = Agentor(
    name="Assistant",
    model="gemini/gemini-2.5-flash",
    api_key=os.environ["GEMINI_API_KEY"],
)

result = agent.run("What are the latest advances in AI?")

Agent with tools

from agentor import Agentor, function_tool

@function_tool
def get_weather(city: str) -> str:
    """Get the current weather for a city.

    Args:
        city: The city to look up.
    """
    return f"The weather in {city} is sunny and 22C."

agent = Agentor(
    name="Weather Agent",
    instructions="Use the weather tool to answer questions.",
    tools=[get_weather],
)

result = agent.run("What's the weather in San Francisco?")

Agent with model settings

from agentor import Agentor, ModelSettings

agent = Agentor(
    name="Creative Writer",
    model="gpt-4o",
    model_settings=ModelSettings(temperature=0.7, max_tokens=1000),
    instructions="You are a creative writing assistant.",
)

Serving an agent

from agentor import Agentor

agent = Agentor(
    name="Customer Support",
    model="gpt-5-mini",
    instructions="You are a helpful customer support agent.",
)

# Serves on http://0.0.0.0:8000
# POST to /chat with {"input": "your message", "stream": false}
agent.serve(port=8000)

Model providers — reach any OpenAI-compatible endpoint
Structured output — return typed objects
Durable runs — save and resume runs
LLM — lightweight LLM client
ModelSettings — configure generation parameters
Tools — create custom tools

Celesto CLI

Celesto API

Celesto CLI

Agentor API

Agentor class API reference

Overview

Constructor

Parameters

Methods

run

arun

resume

fork

chat

stream_chat

serve

from_md

think

RunResult

Usage Examples

Basic agent

Agent with a different provider

Agent with tools

Agent with model settings

Serving an agent

​Overview

​Constructor

​Parameters

​Methods

​run

​arun

​resume

​fork

​chat

​stream_chat

​serve

​from_md

​think

​RunResult

​Usage Examples

​Basic agent

​Agent with a different provider

​Agent with tools

​Agent with model settings

​Serving an agent

​Related

Overview

Constructor

Parameters

Methods

run

arun

resume

fork

chat

stream_chat

serve

from_md

think

RunResult

Usage Examples

Basic agent

Agent with a different provider

Agent with tools

Agent with model settings

Serving an agent

Related