agentlane/ docs
On this page

Default agents

agentlane.harness.agents.DefaultAgent is the high-level local harness entry point for straightforward agent usage.

Use it when you want to:

  1. define an agent with AgentDescriptor,
  2. call run(...) directly, and
  3. optionally observe live model or high-level run events, and
  4. let the framework manage the local runtime and default runner for you.

This is the primary local agent-building interface. Internally it uses the lower-level addressed agentlane.harness.Agent, but from the developer standpoint it is a first-class agent type, not a secondary adapter.

Starting Point For High-Level Agents

DefaultAgent is the recommended starting point when you are developing a high-level local agent. It gives you one direct run(...) surface while still using the same descriptor, tools, shims, runner, and run-state contracts as the lower-level harness.

A practical high-level agent usually starts with:

  1. AgentDescriptor for the model, instructions, tools, shims, and optional explicit agent name
  2. DefaultAgent for local runtime and runner ownership
  3. HarnessToolsShim when the agent needs first-party workspace tools
  4. Tools limits when you want predictable tool-loop behavior

If the descriptor name is omitted or blank, AgentLane generates a Faker-backed adjective-noun fallback name. Applications that care about stable trace labels should pass the name explicitly.

This path keeps early application code compact. If the agent later needs direct runtime addressing, distributed workers, or custom orchestration, move the same descriptor and tool definitions to the lower-level harness APIs. Read Harness Distributed Agents for the pattern that keeps DefaultAgent at the top while worker agents move behind explicit runtimes.

Import Path

from agentlane.harness.agents import AgentBase, DefaultAgent

AgentBase is the abstract base contract for future high-level harness agent types. It defines four shared execution controls:

  1. run(...)
  2. run_stream(...)
  3. fork(...)
  4. reset()

DefaultAgent also exposes run_events(...) for high-level lifecycle streaming without requiring every future AgentBase implementation to support that broader event surface immediately.

Authoring Styles

Subclass-Based

from agentlane.harness import AgentDescriptor
from agentlane.harness.agents import DefaultAgent


class CareNavigationAgent(DefaultAgent):
    descriptor = AgentDescriptor(
        name="Care Navigation",
        model=model,
        instructions="You are a patient care navigation agent.",
    )


agent = CareNavigationAgent()
result = await agent.run("I feel dizzy after starting a new medication.")

Streaming

stream = await agent.run_stream("I feel dizzy after starting a new medication.")

async for event in stream:
    ...

result = await stream.result()

Direct Construction

from agentlane.harness import AgentDescriptor
from agentlane.harness.agents import DefaultAgent

agent = DefaultAgent(
    descriptor=AgentDescriptor(
        name="Portfolio Risk",
        model=model,
        instructions="You are a portfolio risk analyst.",
    )
)

result = await agent.run("Review semiconductor exposure before rebalancing.")

From A Markdown File

DefaultAgent.from_markdown(...) builds the descriptor from a Claude-Code-style AGENT.md file (YAML frontmatter + a system-prompt body) instead of in code. A top-level agent must resolve to a model, supplied by the model= argument or by a model_resolver that builds the frontmatter model spec:

from agentlane.harness.agents import DefaultAgent

agent = DefaultAgent.from_markdown("care_navigator.md", model=model)
result = await agent.run("I feel dizzy after starting a new medication. What first?")

Pass subagents=[...] to attach markdown (or descriptor) sub-agents. See Markdown Agent Definitions for the frontmatter schema, model resolution, tool policy, and sub-agent wiring.

What DefaultAgent Owns

DefaultAgent owns the high-level local agent experience:

  1. descriptor resolution
  2. optional runtime provisioning
  3. optional runner provisioning
  4. persisted RunState between repeated run(...) calls
  5. live streaming through run_stream(...)
  6. high-level lifecycle streaming through run_events(...)
  7. binding and reuse of any configured harness shims
  8. portable committed-state snapshots through snapshot()
  9. atomic local persistence through state_path= or a custom StateStore

It delegates the real orchestration to the existing runtime-facing harness stack:

  1. agentlane.harness.Agent
  2. AgentLifecycle
  3. bound shims, when configured
  4. Runner

Reading The Result

run(...), fork(...), and stream.result() all return a RunResult with four fields:

  1. final_output is the answer extracted from the terminal run turn
  2. responses is the raw ModelResponse list accumulated across the run
  3. turn_count is the number of model turns completed
  4. run_state is the final resumable RunState when available
result = await agent.run("I feel dizzy after starting a new medication.")
print(result.final_output)

Runtime Behavior

If no runtime is supplied:

  1. DefaultAgent provisions a local single-threaded runtime for the call
  2. starts it automatically
  3. drains it and tears it down cleanly on exit

If a runtime is supplied:

  1. DefaultAgent uses that runtime
  2. scopes startup and shutdown to the call only when needed
  3. does not replace the supplied runtime with a different one

Runner Behavior

If no runner is supplied:

  1. DefaultAgent creates one default Runner()
  2. reuses it on later run(...) calls for that agent instance

If a runner is supplied:

  1. DefaultAgent uses it directly
  2. does not replace it with a new default runner

Repeated Runs

Repeated run(...) calls on the same agent instance continue the same conversation by default.

first = await agent.run("I feel dizzy after starting a new medication.")
second = await agent.run("Please summarize the next step.")

The second call continues from the persisted RunState returned by the first call.

One DefaultAgent instance also serializes concurrent run(...) calls on itself so its saved conversation state stays coherent.

run_stream(...) follows that same saved conversation and the same serialization rule. A streamed run commits the updated RunState only after the stream completes successfully.

run_events(...) follows the same state and serialization rule while yielding high-level run events instead of only model stream events.

If the descriptor declares shims, DefaultAgent binds them once for that concrete agent instance and reuses those bound sessions across repeated runs.

Ephemeral messages apply to one run only. A run sends them on each model call, and the committed RunState does not contain them. The next run does not receive them. See Ephemeral context.

Persistent State Path

Pass state_path= to make a DefaultAgent survive process restarts without writing persistence glue:

from agentlane.harness import AgentDescriptor
from agentlane.harness.agents import DefaultAgent

agent = DefaultAgent(
    descriptor=AgentDescriptor(name="Portfolio Risk", model=model),
    state_path=".agentlane/portfolio-risk.json",
)
result = await agent.run("Review semiconductor exposure.")

If the file exists, construction restores its agent address, instructions, conversation, responses, shim state, turn count, and revision. The new prompt continues that state. After a successful primary run, the next committed revision atomically replaces the file. A missing file is created after the first successful run. File read, JSON parsing, and snapshot validation errors propagate to the caller.

run(...), run_stream(...), and run_events(...) all use the same persistence path. fork(...) remains isolated and does not overwrite the primary state. The saved file does not contain ephemeral messages. reset() removes the persisted file. It raises RuntimeError if a primary run is active.

The local file store assumes one live writer per agent. Atomic replacement prevents torn JSON files. On POSIX systems, the store also syncs the file and its parent directory. Revision checks reject stale sequential saves. Applications that allow concurrent writers must coordinate them outside the harness.

StateStore is the public abstract base class for custom storage. Implement load(), save(), and delete(), then pass the store to DefaultAgent:

from agentlane.harness import AgentSnapshot, StateStore


class DatabaseStateStore(StateStore):
    def load(self) -> AgentSnapshot | None:
        ...

    def save(
        self,
        snapshot: AgentSnapshot,
        *,
        expected_revision: int | None,
    ) -> None:
        ...

    def delete(self, *, expected_revision: int | None) -> None:
        ...


agent = DefaultAgent(
    descriptor=AgentDescriptor(name="Portfolio Risk", model=model),
    state_store=database_state_store,
)

Each store instance manages one agent snapshot. The save() and delete() methods must check expected_revision. When expected_revision is not None, save() must also require a greater snapshot revision. These checks reject stale or non-increasing state changes. Use state_path= when the built-in JsonFileStateStore is sufficient.

Portable Snapshots

DefaultAgent.snapshot() returns an AgentSnapshot for the latest completed run. It returns None before the first run completes. A snapshot requested while a run is active contains the previous committed state.

from agentlane.harness import AgentDescriptor, AgentSnapshot
from agentlane.harness.agents import DefaultAgent
from agentlane.messaging import AgentId

agent_id = AgentId.from_values("portfolio-risk", "main")
agent = DefaultAgent(
    descriptor=AgentDescriptor(name="Portfolio Risk", model=model),
    agent_id=agent_id,
)
await agent.run("Review semiconductor exposure.")

snapshot = agent.snapshot()
if snapshot is None:
    raise RuntimeError("the agent has no completed run")

payload = snapshot.to_json()
restored_snapshot = AgentSnapshot.from_json(payload)
restored = DefaultAgent(
    descriptor=AgentDescriptor(name="Portfolio Risk", model=model),
    snapshot=restored_snapshot,
)
result = await restored.run("Now summarize the main risk.")

The versioned JSON value contains the stable AgentId, revision, creation time, rendered instructions, canonical model-ready history, raw model responses, JSON-safe shim state, and turn count. Prompt templates are stored as rendered content. Unsupported schema versions and non-JSON shim values raise errors.

Snapshots never contain ephemeral messages. AgentSnapshot.capture(...) removes them from the state that it captures, and persistent internal messages stay. A restored agent can send new run-scoped context with its next input:

from agentlane.models import internal_message

view = internal_message(
    {"role": "user", "content": "The user views the risk chart."},
    origin="ui_view",
    ephemeral=True,
)
result = await restored.run(["Explain the largest risk.", view])

The model receives view during this run only. The next snapshot does not contain it.

The restoring process supplies the live descriptor, model, tools, and shims. The snapshot stores committed run state and logical identity. Use JsonFileStateStore for the local default, implement StateStore for another storage system, or persist the value returned by to_json() in application-owned storage.

Streaming Semantics

run_stream(...) returns a RunStream.

Use it like this:

  1. iterate the stream for live ModelStreamEvent values
  2. await stream.result() for the final RunResult

Important details:

  1. ModelStreamEventKind.COMPLETED is model-call completion, not whole-run completion
  2. one streamed harness run may emit more than one COMPLETED event because of tools or first-class handoff
  3. early close or cancellation does not commit partial RunState
  4. first-class handoff continues the outer stream because control transfers
  5. agent-as-tool stays internal in the current streaming contract

With a Responses model configured for WebSocket use, call await stream.steer(input) while a native response is active. The returned receipt separates queued input from input applied to a successor response. Both run_stream(...) and run_events(...) expose this control. For configuration, async function execution, and receipt states, see Steering and async tools.

Run Events

run_events(...) returns a RunEventStream. It is the broader streaming surface for applications that need to drive dashboards, logs, or debug UIs from one ordered event feed.

Use it like this:

from agentlane.harness import RunEventKind

stream = await agent.run_events("Plan the documentation update.")

async for event in stream:
    if event.kind == RunEventKind.TOOL_START:
        ...

result = await stream.result()

The stream includes:

  1. wrapped ModelStreamEvent values, preserving the existing model stream payload
  2. agent start and end events
  3. LLM request start and end events
  4. tool call start and end events
  5. brokered approval requested/resolved events when approval_events= is provided
  6. first-class handoff start and end events
  7. compact full RunState snapshots at stable boundaries

State snapshots are emitted at run start, after each prepared turn, after each tool round, and at run end. Each snapshot contains the current turn count, history length, response count, and a copied shim_state. It intentionally does not include full conversation history or raw responses.

run_stream(...) remains unchanged and continues to yield only ModelStreamEvent values. Use run_events(...) only when the caller wants the broader lifecycle feed.

Pass approval_events=broker.events() when a host wants approval requested/resolved events from ToolApprovalBroker merged into the same ordered stream.

Fork

DefaultAgent.fork(...) runs one branch without mutating the agent’s persisted primary conversation line.

Simple current behavior:

  1. snapshot the current persisted baseline
  2. run the branch under a fresh runtime agent id
  3. return the branch RunResult
  4. keep DefaultAgent.run_state unchanged

Example:

await agent.run("Review semiconductor exposure before rebalancing.")
branch = await agent.fork("Draft a more formal risk note.")

After that call:

  1. branch.run_state contains the branch result
  2. agent.run_state still points at the original main conversation line

Explicit Resume

DefaultAgent.run(...) also accepts RunState directly.

That path is an explicit resume request. The agent does not combine the passed RunState with its already saved conversation state. It resumes from the provided state only, then stores the returned updated state afterward. When a state store is configured, the stored revision remains the durable revision baseline for that commit.

DefaultAgent.fork(...) also accepts RunState. That creates a branch from the provided state and still does not write the branch result back into the agent’s stored primary state.

Reset

DefaultAgent.reset() clears the agent’s saved RunState.

Use it when the same agent instance should start a new conversation instead of continuing the previous one. It raises RuntimeError while a primary run is active.

Boundary

DefaultAgent is local and convenience-oriented.

agentlane.harness.Agent is the lower-level addressed harness agent.

Use:

  1. DefaultAgent for the smallest local developer surface
  2. Agent when you want direct runtime addressing, explicit binding, or lower level orchestration control
  3. Harness Shims when you need to extend instructions, tools, or turn context without changing the core harness types
  4. Harness Distributed Agents when a top-level harness agent coordinates worker runtimes