Agents

Agents are the core runtime loop in the Python SDK. An Agent is a Pydantic model that coordinates generations, tool calls, and lifecycle events. Every run is captured in a Trajectory, which you can inspect for messages, events, and usage.

Caution

The SDK is pre-1.0. APIs may change between releases.

Configuration fields

Field	Type	Default	Description
name	str	(required)	Display name for logs and tracing.
description	str	""	Brief description of the agent.
model	str | Generator | None	None	Model identifier (e.g. "gpt-4o-mini") or a Generator instance.
instructions	str | None	None	System prompt injected into each run. Automatically dedented.
tools	list[Tool | Toolset]	[]	Tools the agent can use. Accepts callables (auto-converted) and nested lists (auto-flattened).
tool_mode	ToolMode	"auto"	How tool calls are parsed. See Tool modes.
hooks	list[Hook]	[]	Hooks applied during execution. See Hooks and reactions.
stop_conditions	list[StopCondition]	[]	Legacy conditions evaluated against trajectory.steps. Prefer stopping hooks.
judge	Judge | None	None	Optional judge for scoring trajectories. Available for external evaluation workflows.
max_steps	int	10	Maximum generation/tool steps before the agent stops.
generation_timeout	int | None	None	Timeout in seconds for each LLM generation call. None means no timeout.
generate_params_extra	dict	{}	Extra parameters merged into every generation request.
working_dir	Path | None	None	Working directory for tool output offloading and IO. Falls back to Path.cwd().
cache	CacheMode | None	None	Cache behavior for generated messages.
tags	list[str]	[]	Tags for filtering and identification. "agent" is auto-inserted.
label	str | None	None	Optional label for UI and tracing.
agent_id	UUID	auto	Unique identifier for the agent instance.

Create and run an agent

The two primary execution methods are:

• await agent.run(goal) — returns a Trajectory.
• async with agent.stream(goal) as events — yields structured events.

from dreadnode.agents import Agent, tool


@tool
def lookup(question: str) -> str:
    """Fetch an answer from a local map."""
    return f"Local answer for: {question}"


async def main() -> None:
    agent = Agent(
        name="support-agent",
        model="gpt-4o-mini",
        instructions="Answer concisely and cite tools when used.",
        tools=[lookup],
        max_steps=25,
    )

    trajectory = await agent.run("What is Dreadnode?")
    print(trajectory.messages[-1].content)
    print(trajectory.usage)

Stream events

Use stream() when you want to observe lifecycle events as they happen.

from dreadnode.agents import Agent


async def main() -> None:
    agent = Agent(name="streamer", model="gpt-4o-mini")

    async with agent.stream("Summarize the platform.") as events:
        async for event in events:
            print(type(event).__name__, event.status)

Shared trajectories

Pass trajectory= to share state between agents or reset=False to continue a conversation without clearing history.

from dreadnode.agents import Agent
from dreadnode.agents.trajectory import Trajectory

shared = Trajectory()

async with agent_a.stream("Research the topic", trajectory=shared) as s:
    async for _ in s:
        pass

# agent_b picks up where agent_a left off
async with agent_b.stream("Summarize the findings", trajectory=shared) as s:
    async for _ in s:
        pass

Tool modes

tool_mode controls how tool calls are parsed from model output.

Mode	Description
"auto"	Default. Uses "api" if the generator supports function calling, otherwise falls back to "json-in-xml".
"api"	Delegates to the provider’s native function calling API.
"xml"	Tool calls parsed in nested XML format. Adds a tool stop token.
"json"	Tool calls parsed as raw name/argument JSON in assistant content.
"json-in-xml"	JSON arguments inside an XML envelope.
"json-with-tag"	JSON structures inside an XML tag.
"pythonic"	Tool calls parsed as Python function call syntax (e.g. tool_name(arg=value)).

Use "api" with mainstream providers (OpenAI, Anthropic). Use "xml" or "json-in-xml" for open-source models that do not support native function calling.

Extended thinking

Use generate_params_extra to enable extended thinking for models that support it.

agent = Agent(
    name="deep-thinker",
    model="claude-sonnet-4-20250514",
    generate_params_extra={
        "thinking": {"type": "enabled", "budget_tokens": 8000},
    },
)

Agent as evaluation task

Use agent.task() to convert an agent into a Task for use with Evaluation or Study. The task takes a goal: str and returns a Trajectory.

from dreadnode.agents import Agent

agent = Agent(name="eval-target", model="gpt-4o-mini")

# Use in an evaluation
@dn.evaluation(dataset=[{"goal": "Explain TLS"}])
async def eval_agent(goal: str) -> Trajectory:
    return await agent.run(goal)

# Or convert directly
task = agent.task(name="agent-task")

Hooks and reactions

Hooks are callables that receive an AgentEvent and can return a reaction to steer execution. Use the @hook decorator to filter by event type.

Reactions

When a hook returns a reaction, it controls what happens next. If multiple hooks react to the same event, priority determines which wins.

Reaction	Priority	Effect
Finish(reason?)	Highest	Stop execution successfully.
Fail(error)	High	Stop execution with an error.
Retry(messages?)	Medium	Retry the step, optionally replacing messages.
RetryWithFeedback(feedback)	Medium	Retry with a feedback message injected.
Continue(messages?, feedback?)	Low	Continue execution, optionally injecting messages or feedback.

Hookable events

Event	When emitted
AgentStart	Agent execution begins.
AgentEnd	Agent execution ends (has stop_reason, error).
AgentStalled	No tool calls and no stop conditions met.
AgentError	Unrecoverable error occurred.
GenerationStart	Before an LLM call.
GenerationEnd	After an LLM call completes.
GenerationStep	Full generation step for trajectory.
GenerationError	Error during generation.
ToolStart	Before a tool call executes.
ToolEnd	After a tool call completes (has result, output_file).
ToolStep	Tool result for trajectory.
ToolError	Error during tool execution.
ReactStep	A hook returned a reaction.
UserInputRequired	Agent needs human input.
Heartbeat	Keepalive during long operations.

Hook example

from dreadnode.agents import Agent, Finish, RetryWithFeedback
from dreadnode.agents.events import GenerationStep, ToolError
from dreadnode.core.hook import hook


@hook(GenerationStep)
def stop_if_short(event: GenerationStep) -> Finish | None:
    if event.messages and len(event.messages[-1].content or "") < 20:
        return Finish("Response too short")
    return None


@hook(ToolError)
def retry_on_tool_error(event: ToolError) -> RetryWithFeedback:
    return RetryWithFeedback(f"Tool failed: {event.error}. Try a different approach.")


agent = Agent(
    name="quality-agent",
    model="gpt-4o-mini",
    hooks=[stop_if_short, retry_on_tool_error],
)

Built-in hooks

Hook	Description
backoff_on_ratelimit	Exponential backoff on rate limit errors (max 8 retries, 300s). Requires litellm.
tool_metrics(detailed=False)	Logs metrics about tool usage, execution time, and success rates. detailed=True logs per-tool metrics.
summarize_when_long	Auto-summarizes conversation when context exceeds 100K tokens or a context-length error occurs.

default_hooks() returns [tool_metrics(), summarize_when_long, step_count(50)] plus backoff_on_ratelimit if available. These are applied automatically when agents run via the server.

Stopping hooks

Stopping hooks are hook factories that return Finish reactions when conditions are met. These are the preferred way to control when an agent stops.

from dreadnode.agents import Agent
from dreadnode.agents.stopping import (
    step_count,
    token_usage,
    elapsed_time,
    tool_use,
    output,
)

agent = Agent(
    name="bounded-agent",
    model="gpt-4o-mini",
    hooks=[
        step_count(20),
        token_usage(50_000),
        elapsed_time(120),
    ],
)

Factory	Listens to	Description
step_count(max_steps)	GenerationStep	Stop after N steps.
generation_count(max)	AgentEvent	Stop after N LLM inference calls (counts retries).
tool_use(tool_name, count=1)	AgentEvent	Stop after a specific tool is used N times.
any_tool_use(count=1)	AgentEvent	Stop after any tool is used N total times.
output(pattern, ...)	GenerationEnd	Stop if pattern found in generated text. Supports case_sensitive, exact, regex.
tool_output(pattern, tool_name?)	ToolEnd	Stop if pattern found in tool output.
tool_error(tool_name?)	ToolError	Stop on tool error.
no_new_tool_used(for_steps)	AgentEvent	Stop if no previously-unseen tool is used for N steps.
no_tool_calls(for_steps=1)	AgentEvent	Stop if no tool calls for N consecutive steps.
token_usage(limit, mode="total")	GenerationEnd	Stop if token usage exceeds limit. Mode: "total", "in", "out".
elapsed_time(max_seconds)	AgentEvent	Stop if wall-clock time exceeds limit.
estimated_cost(limit)	GenerationEnd	Stop if estimated LLM cost exceeds USD limit.
consecutive_errors(count)	AgentEvent	Stop after N consecutive tool errors.

Trajectory

Trajectory collects every event and message in order. Key properties:

• trajectory.events — raw events list.
• trajectory.messages — reconstructed conversation history.
• trajectory.usage — token usage totals.
• trajectory.steps — step events for stop condition evaluation.

Reset and continuation

• agent.reset() — clears internal state and returns the previous trajectory.
• stream(reset=False) — continues the conversation without clearing history.
• stream(trajectory=shared) — uses an external trajectory object.