jeevesagent.agent.api

The public Agent class.

Conventions:

  • Pass a string of instructions for a working agent backed by sensible defaults: EchoModel, InMemoryMemory, InProcRuntime, NoBudget, AllowAll, HookRegistry, an empty InProcessToolHost.

  • Pass tools=[fn_or_Tool, ...] to register Python callables; the agent wraps them in an in-process ToolHost.

  • Override any subsystem by passing a concrete implementation of the matching protocol from jeevesagent.core.protocols.

Two execution surfaces share a single internal loop:

  • Agent.run() runs to completion and returns a RunResult.

  • Agent.stream() returns an AsyncIterator[Event] of milestones as they happen — STARTED, MODEL_CHUNK, TOOL_CALL, TOOL_RESULT, BUDGET_WARNING/EXCEEDED, ERROR, COMPLETED.

Internally, _loop() accepts an emit callback and threads it through every milestone. run() passes a no-op emit; stream() pipes events through an anyio.create_memory_object_stream() so a slow consumer applies backpressure to the loop instead of buffering unboundedly.

Attributes

DEFAULT_MAX_TURNS

DEFAULT_STREAM_BUFFER

Emit

Classes

Agent

A fully-async, MCP-native, model-agnostic agent harness.

Module Contents

class jeevesagent.agent.api.Agent(instructions: str, *, model: jeevesagent.core.protocols.Model | str | None = None, memory: jeevesagent.core.protocols.Memory | None = None, runtime: jeevesagent.core.protocols.Runtime | None = None, budget: jeevesagent.core.protocols.Budget | None = None, permissions: jeevesagent.core.protocols.Permissions | None = None, hooks: jeevesagent.security.hooks.HookRegistry | None = None, tools: list[jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]] | jeevesagent.core.protocols.ToolHost | jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object] | None = None, telemetry: jeevesagent.core.protocols.Telemetry | None = None, audit_log: jeevesagent.security.audit.AuditLog | None = None, max_turns: int = DEFAULT_MAX_TURNS, auto_consolidate: bool = False, architecture: jeevesagent.architecture.Architecture | str | None = None, skills: list[Any] | None = None, retry_policy: jeevesagent.governance.retry.RetryPolicy | None = None)[source]

A fully-async, MCP-native, model-agnostic agent harness.

add_tool(item: jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]) → jeevesagent.tools.registry.Tool[source]

Register a tool after construction.

Convenience for plugin-style code that adds tools after the Agent exists. Only works when the underlying tool host is an InProcessToolHost (the default — and the only host that has a writable registry today).

Returns the constructed Tool so callers can introspect the auto-derived schema.

after_tool(fn: jeevesagent.security.hooks.PostToolHook) → jeevesagent.security.hooks.PostToolHook[source]

Register a best-effort post-tool callback.

before_tool(fn: jeevesagent.security.hooks.PreToolHook) → jeevesagent.security.hooks.PreToolHook[source]

Register a pre-tool hook. First denial wins; allow otherwise.

async consolidate() → int[source]

Manually trigger memory consolidation.

Returns the number of new facts the consolidator extracted, or 0 when the memory backend doesn’t expose a fact store.

Useful when auto_consolidate=False (the default) and you want to batch consolidation at a controlled cadence — e.g. once a day, or before shutdown.

classmethod from_config(path: str | pathlib.Path, *, model: jeevesagent.core.protocols.Model | None = None, memory: jeevesagent.core.protocols.Memory | None = None, runtime: jeevesagent.core.protocols.Runtime | None = None, tools: list[jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]] | jeevesagent.core.protocols.ToolHost | None = None) → Agent[source]

Construct an Agent from a TOML config file.

Designed for ops/devops users who want declarative agent config separate from code. Supports the textual / numeric bits — instructions, model spec (string), max_turns, auto_consolidate, budget — and lets callers pass concrete instances for the things TOML can’t reasonably express (real Memory, Runtime, custom Model, tools).

Example agent.toml:

instructions = "You are a research assistant."
model = "claude-opus-4-7"
max_turns = 100
auto_consolidate = true

[budget]
max_tokens = 200_000
max_cost_usd = 5.0
max_wall_clock_minutes = 10
soft_warning_at = 0.8

Then:

agent = Agent.from_config("agent.toml")
classmethod from_dict(cfg: dict[str, Any], *, model: jeevesagent.core.protocols.Model | None = None, memory: jeevesagent.core.protocols.Memory | None = None, runtime: jeevesagent.core.protocols.Runtime | None = None, tools: list[jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]] | jeevesagent.core.protocols.ToolHost | None = None) → Agent[source]

Construct an Agent from a parsed config dict.

Same shape as from_config() but skips the file read. Useful when the config comes from somewhere other than a TOML file — environment variables, a Pydantic settings model, a yaml.safe_load result, an HTTP API, etc.

Recognised keys (all optional except instructions and model):

  • instructions: str — required

  • model: str — required (or pass model= kwarg)

  • max_turns: int

  • auto_consolidate: bool

  • budget: dict with any of max_tokens, max_input_tokens, max_output_tokens, max_cost_usd, max_wall_clock_minutes, soft_warning_at

async generate_graph(path: str | pathlib.Path | None = None, *, title: str | None = None) → str[source]

Render this agent’s structure as a Mermaid graph.

Walks the agent + its architecture + all sub-agents + every agent’s tools, producing a graph that captures the full team, tool attachments, and architecture-specific relationships (delegate / handoff / classify / etc.).

Returns the Mermaid text. If path is provided, also writes to disk — extension determines the format:

  • .mmd — raw Mermaid source

  • .md — Markdown with the diagram in a mermaid fence (renders on GitHub, IDE markdown previews, Jupyter)

  • .png / .svg — rendered via mermaid.ink; falls back to .mmd next to the path on network failure

Example:

mermaid_text = await agent.generate_graph("graph.md")
print(mermaid_text)

Pass title= to override the diagram title (defaults to the file’s stem, or "Agent" if no path is given).

async recall(query: str, *, kind: str = 'episodic', limit: int = 5) → list[Any][source]

Convenience wrapper around self.memory.recall(query, ...).

Returns episodes matching query. For semantic / fact-store recall, use self.memory.facts.recall_text(...) directly.

remove_tool(name: str) → bool[source]

Unregister a tool by name. Returns True if a tool was removed, False if no tool with that name was registered.

Same constraint as add_tool(): only works with InProcessToolHost.

async resume(session_id: str, prompt: str, *, user_id: str | None = None, metadata: collections.abc.Mapping[str, Any] | None = None, context: jeevesagent.core.context.RunContext | None = None, extra_tools: list[jeevesagent.tools.registry.Tool] | None = None, emit: Emit | None = None, output_schema: type[pydantic.BaseModel] | None = None, output_validation_retries: int = 1) → jeevesagent.core.types.RunResult[source]

Resume a previously-interrupted run from its journal.

Equivalent to agent.run(prompt, session_id=session_id, ...) with the same kwarg surface as run(). Exists as a separate method so the intent is explicit at the call site — when a durable Runtime (e.g. SqliteRuntime) is configured, completed steps replay from the journal instead of re-executing.

async run(prompt: str, *, user_id: str | None = None, session_id: str | None = None, metadata: collections.abc.Mapping[str, Any] | None = None, context: jeevesagent.core.context.RunContext | None = None, extra_tools: list[jeevesagent.tools.registry.Tool] | None = None, emit: Emit | None = None, output_schema: type[pydantic.BaseModel] | None = None, output_validation_retries: int = 1) → jeevesagent.core.types.RunResult[source]

Run the agent to completion and return its RunResult.

user_id is the namespace partition for memory recall and persistence — episodes and facts stored with one user_id are never visible to a query scoped to a different user_id. None is the “anonymous / single-tenant” bucket. See RunContext for the partitioning contract.

Pass session_id to resume a journaled run — when paired with a durable runtime (e.g. SqliteRuntime), already-completed steps replay from the journal instead of re-executing. Without a durable runtime, session_id just labels the run.

metadata is a free-form bag for application context the framework does not interpret (locale, request id, feature flags). Tools and hooks read it via get_run_context().metadata.

context accepts a fully-formed RunContext instead of the individual kwargs — useful when passing context through multi-agent boundaries that received their parent’s context as a single object. When both context and the individual kwargs are provided, the kwargs override the corresponding fields on context.

extra_tools injects additional Tools for this run only — the agent’s configured ToolHost is wrapped so the model sees the extras alongside whatever tools were registered at construction. Used by multi-agent architectures that need to inject coordination tools (e.g. Swarm’s handoff(target, message)) into a peer agent’s loop without permanently mutating that agent’s static configuration.

emit is an awaitable callback invoked once per Event produced during the run (model chunks, tool calls, tool results, architecture progress, errors, …). Default None drops events on the floor (regular run semantics — return only the final RunResult). Multi-agent architectures pass an emit that forwards a sub-Agent’s events into the parent’s stream, so calls like await worker.run(prompt, emit=parent_send) surface the worker’s token-by-token streaming to the outermost agent.stream(...) consumer.

output_schema requests a structured, validated final answer. Pass any Pydantic BaseModel subclass and the framework will (1) append a JSON-schema directive to the system prompt instructing the model to emit a final answer that matches, (2) parse the final assistant text against the schema, and (3) populate RunResult.parsed with the validated instance. RunResult.output keeps the raw text so you can log or display it. Up to output_validation_retries extra turns are spent recovering from a parse failure (the model is given the validation error as feedback and asked to try again); if it still fails after the retry budget, the run raises OutputValidationError. Set retries to 0 to fail fast.

async stream(prompt: str, *, user_id: str | None = None, session_id: str | None = None, metadata: collections.abc.Mapping[str, Any] | None = None, context: jeevesagent.core.context.RunContext | None = None, extra_tools: list[jeevesagent.tools.registry.Tool] | None = None, output_schema: type[pydantic.BaseModel] | None = None, output_validation_retries: int = 1) → collections.abc.AsyncIterator[jeevesagent.core.types.Event][source]

Stream Events as the loop produces them.

The loop runs as a background task; events are pushed through a bounded memory stream so a slow consumer applies backpressure. Breaking out of the iteration cancels the producer cleanly. session_id works the same as run()’s — pass an existing one to resume against a durable runtime’s journal. extra_tools works the same as run()’s.

async tools_list() → list[str][source]

Return the names of all currently-registered tools.

Convenience that works for any ToolHost. Calls tool_host.list_tools() under the hood and returns just the names; use self.tool_host.list_tools() directly for the full ToolDef records.

with_tool(fn: collections.abc.Callable[Ellipsis, object]) → collections.abc.Callable[Ellipsis, object][source]

Decorator-style equivalent of add_tool().

Usage:

@agent.with_tool
async def search(query: str) -> str:
    '''Search a knowledge base.'''
    return f"results for {query}"

Returns the original function unchanged (so it can still be called normally), and registers it as a tool on the agent’s underlying InProcessToolHost. Same constraint as add_tool(): the host must be writable.

property architecture: jeevesagent.architecture.Architecture

The configured Architecture strategy.

Default is ReAct. Pass architecture= to Agent(...) to override.

property budget: jeevesagent.core.protocols.Budget

The configured Budget.

property hooks: jeevesagent.core.protocols.HookHost
property instructions: str

The system prompt the agent runs with.

Surfaced as a public property so multi-agent architectures (e.g. Supervisor) can read each worker’s intended role when composing instructions for the supervising model.

property memory: jeevesagent.core.protocols.Memory

The configured Memory backend.

property model: jeevesagent.core.protocols.Model

The configured Model adapter.

property permissions: jeevesagent.core.protocols.Permissions

The configured Permissions policy.

property runtime: jeevesagent.core.protocols.Runtime

The configured Runtime.

property skills: Any | None

The SkillRegistry of skills registered on this agent (or None if no skills were configured). Useful for inspecting / mutating the skill set after construction.

property tool_host: jeevesagent.core.protocols.ToolHost

The configured ToolHost.

jeevesagent.agent.api.DEFAULT_MAX_TURNS = 50
jeevesagent.agent.api.DEFAULT_STREAM_BUFFER = 128
jeevesagent.agent.api.Emit