Running agents¶
How to run agents: one-off agent.run(), streaming, and using the Runtime directly.
Using Agent.run()¶
The simplest way to run an agent is to create an Agent and call run():
from routekitai import Agent
from routekitai.providers.local import FakeModel
model = FakeModel(name="test")
model.add_response("Done.")
agent = Agent(name="my_agent", model=model, tools=[])
result = await agent.run("Do something")
# result.output — final Message
# result.trace_id — ID for this run
# result.messages — full conversation
# result.final_state — agent state at end
- The agent creates an internal
Runtimeand registers itself, so you don’t need to manage a runtime for basic use. - Traces are written to
.routekit/traces/by default. To use another directory, passtrace_dirwhen creating the agent:
python
from pathlib import Path
agent = Agent(name="my_agent", model=model, tools=[], trace_dir=Path("/tmp/my_traces"))
Streaming: agent.run_stream()¶
For real-time updates (e.g. progress or trace events), use run_stream():
async for event in agent.run_stream("Long task"):
if event["type"] == "trace_event":
print("Event:", event["data"].get("type"))
elif event["type"] == "progress_update":
print("Progress:", event["data"].get("progress_percent"), "%")
elif event["type"] == "result":
result = event["result"]
print("Output:", result.output.content)
Events include:
trace_event— Raw trace event (e.g. step_started, tool_called).progress_update— Progress info (e.g.progress_percent).result— FinalRunResult(only in the last event).
Using Runtime directly¶
When you need multiple agents, shared hooks, or a custom policy (e.g. graph, supervisor), use a Runtime and register agents:
from pathlib import Path
from routekitai.core.runtime import Runtime
from routekitai.core.agent import Agent
runtime = Runtime(trace_dir=Path(".routekit/traces"))
runtime.register_agent(agent1)
runtime.register_agent(agent2)
result = await runtime.run("agent1", "Prompt", policy=some_policy)
runtime.run(agent_name, prompt, policy=..., **kwargs)— Runs the named agent with the given prompt and optional policy.runtime.replay(trace_id, agent_name)— Replays a trace (requirestrace_dirset). See Tracing and replay for details.
You can pass policy hooks (PII redaction, tool filter, approval gate) on the runtime so they apply to all runs:
from routekitai.core.hooks import PolicyHooks, PIIRedactionHook
runtime = Runtime(
trace_dir=Path(".routekit/traces"),
policy_hooks=PolicyHooks(pii_redaction=PIIRedactionHook(redact_emails=True)),
)
Policies¶
By default, an agent uses a ReAct-style policy (reasoning and acting in a loop). You can pass a different policy when creating the agent or when calling runtime.run():
- ReActPolicy — Model can call tools in a loop until it returns a final answer.
- FunctionCallingPolicy — Strict function-calling mode.
- GraphPolicy — Execute a graph workflow; see Graph workflows.
- SupervisorPolicy — One agent delegates to others.
- PlanExecutePolicy — Plan first, then execute.
Example with a custom policy on the agent:
from routekitai.core.policies import ReActPolicy
agent = Agent(
name="my_agent",
model=model,
tools=[...],
policy=ReActPolicy(max_iterations=15),
)
Timeouts and concurrency¶
- Runtime accepts
timeout(default timeout for a run) andmax_concurrency(parallel tool calls). - Tool can set
timeoutfor that tool’s execution.
See Security & Governance for timeouts, retries, and approval gates.