| name | continuum-agent |
| description | Build BaseAgent instances and run them with AgentRunner — covers fields, lifecycle hooks, structured outputs, ReAct mode, instruction modifiers, and the full execution flow. Invoke when the user asks "create an agent", "configure max_turns", "add lifecycle hooks", "structured output with Pydantic", or anything around the core agent abstraction. |
Continuum Agent Skill
Authoritative source: docs/agent.md.
Core API
from orchestrator.agent import BaseAgent, AgentRunner
from orchestrator.agent.config import AgentConfig, AgentMemoryConfig, RunnerConfig
from orchestrator.agent.types import (
Handoff, MemoryScope, RunContext, AgentResponse, EventType,
)
Minimal agent
agent = BaseAgent(
name="my-agent",
instructions="...",
model="gpt-4o-mini",
temperature=0.7,
max_tokens=None,
)
runner = AgentRunner()
resp = await runner.run(agent, "user input", user_id="u1", session_id="s1")
print(resp.content)
print(resp.usage.total_tokens)
print(resp.structured_output)
Important BaseAgent parameters
| Parameter | Default | Notes |
|---|
name | required | unique id |
instructions | "" | supports {template_vars} |
model | settings.default_llm_model | prefix routes provider |
tools | [] | LLM-shaped dicts |
mcp_servers | [] | Auto-discovered tool sources |
handoffs | [] | [Handoff(target_agent=..., description=...)] |
memory_config | AgentMemoryConfig() | search_memories=True / store_memories=True |
output_schema | None | Pydantic model |
template_vars | {} | static slots |
examples | [] | few-shot |
instruction_modifiers | [] | dynamic prompt rewriters |
on_start / on_end / on_error / on_tool_call / on_handoff | None | sync callables |
config | AgentConfig() | max_turns=25, react_mode, reasoning_mode, scanners, … |
gateway_mode | None | Smart Gateway routing: "strict" / "modest" / "quality" |
policy_store | None | Security policy store for request validation |
Structured output
from pydantic import BaseModel
class Plan(BaseModel):
intent: str
steps: list[str]
agent = BaseAgent(name="planner", instructions="...", output_schema=Plan)
resp = await runner.run(agent, "...")
plan: Plan = resp.structured_output
Streaming
from orchestrator.agent.types import EventType
async for ev in runner.run_stream(agent, "..."):
if ev.type == EventType.CONTENT_DELTA:
print(ev.data["content"], end="", flush=True)
Memory scope (the agent enum, not the dataclass!)
from orchestrator.agent.types import MemoryScope
ReAct + reasoning modes
agent = BaseAgent(
name="thinker",
instructions="...",
config=AgentConfig(react_mode=True, reasoning_mode=False),
)
react_mode injects a hidden think tool the LLM must call before
producing a final answer. reasoning_mode does a silent "think first"
pass before the main loop.
RAG context
agent = BaseAgent(
name="rag-agent",
instructions="Answer using PROVIDED CONTEXT.",
config=AgentConfig(rag_context=retrieved_docs, require_context=True),
)
Hooks
def on_tool_call(agent, tool_name, args):
print(f"{agent.name} -> {tool_name}({args})")
agent = BaseAgent(name="...", instructions="...", on_tool_call=on_tool_call)
Hooks are sync. Async hooks are not awaited.
Execution flow (what runner.run() does)
- Validate input against
agent.input_schema (if any)
- Build
RunContext and RunState; start trace
- Load tool-context state from session
- Assemble messages: system prompt → ReAct scaffold → tool context →
memory facts (Qdrant) → session history (Redis) → optional RAG →
sanitized user input → optional context compression
- Optional reasoning pass (silent)
- Loop up to
max_turns: LLM call → run tools / handoffs → feed results back
- Persist session, store memories, return
AgentResponse
Don't
- Don't pass
role= / content= to SessionClient.add_message — pass
a ChatMessage object.
- Don't import
MemoryScope from orchestrator.memory and then pass
it to AgentMemoryConfig — that's the dataclass; the agent module
re-exports the enum of the same name.
- Don't make hooks async — they're sync callables.
- Don't change
max_turns to a huge number; if a loop fails to
converge in 25 turns the prompt is the problem.