OmniCoreAgent

from omnicoreagent import OmniCoreAgent

Constructor

Argument	Default	What it is
`name`	`required`	The agent’s name, in its traces, records and workspace.
`system_instruction`	`required`	What the agent is for; the start of its system prompt, before the runtime’s own instructions.
`model_config`	`required`	The model: `{"provider": "openai", "model": "..."}`, with optional `temperature`, `max_tokens`, `top_p`, `reasoning_effort`, `logprobs` and `api_key`, and `azure_*` or `ollama_host` for those providers (see Models). The key comes from `LLM_API_KEY` unless `api_key` is given.
`mcp_tools`	`None`	MCP servers whose tools the agent may use: a list of dicts with `name`, `transport_type` (`stdio`, `sse` or `streamable_http`), and `command`/`args` or `url`.
`local_tools`	`None`	Your Python functions as tools: a `ToolRegistry` (functions registered with `@tools.register_tool`), or a list of `Tool` objects.
`sub_agents`	`None`	Other `OmniCoreAgent` instances this one may hand a task to: each becomes a `delegate_<name>` tool.
`agent_config`	`None`	The agent’s settings (see Agent settings).
`memory_router`	`None`	Where session history is kept: `MemoryRouter(“in_memory"	"redis"	"sql"	"mongodb”)`; in memory by default.
`telemetry_store`	`None`	A trace store to use instead of the built-in one, for example one shared by several agents.
`telemetry_recorder`	`None`	A recorder to use instead of the built-in one.
`telemetry_stream`	`None`	A stream of telemetry events to publish to.
`telemetry_exporters`	`None`	Where traces are exported as they finish: OTLP, LangSmith, Opik or JSONL exporters.
`prompt_builder`	`None`	Replaces how the system prompt is assembled.
`debug`	`False`	Log each step in detail.
`telemetry_config`	`None`	What a trace records, where, and for how long (see Telemetry settings).
`telemetry_payload_store`	`None`	Where payloads too large to keep inline are stored, when offloading is on.

agent_config takes the agent settings, and telemetry_config the telemetry settings.

Methods

`abandon_run`

async def abandon_run(run_id: str, *, status: str, reason: str) -> Dict[str, Any] | None: ...

Close a run this agent did not finish itself. For whoever ended it from outside: a background run cancelled while it waited, or failed because its worker died. The record says status (cancelled, failed or timeout) and why, and the request’s own budget counters are released, keeping what it spent on the record. A run that already ended is left as it is. Found on the steward’s server: such runs stayed “running” or “awaiting_budget” for good, each with its counter still in the ledger.

`budget_status`

async def budget_status(run_id: str) -> List[Dict[str, Any]]: ...

Every budget covering a run: its limit, what a person granted on top of it, what is spent and held, and what remains. The application’s, session’s and agent’s counters are read from the ledger (they are shared, and live on); a finished run’s own counter is the one settled on its record. Empty when nothing is budgeted.

`claim_orphaned_runs`

async def claim_orphaned_runs(*, limit: int = 10, max_recoveries: int = 3, decided_grace_seconds: float = 30.0) -> List[Dict[str, Any]]: ...

Take over this agent’s runs that nobody is working on, and say which. Two kinds of run are taken. An orphan is a run still running whose heartbeat is older than its lease, started by this agent (by name) and not by a background supervisor, which recovers its own. A decided run is one waiting for an approval or a budget decision that a person has made, and that nobody has resumed for decided_grace_seconds (the client that should have called resume was refused or is gone). A run with a decision still pending is left waiting. Each run is claimed with the record’s own version check: the claim writes a new owner and a fresh heartbeat, and a second process that read the same record fails the check and skips it, so a run is never resumed twice. A client’s resume that read the record before the claim fails its first save the same way and is refused. Each claim is then given to resume_claimed. An orphan claimed max_recoveries times and still not finished is ended failed instead (it is probably what kills its process); a decided run’s claim does not count. Orphans were added after the support desk chaos run (2026-10-07), where runs whose process died stayed running until someone called resume. Decided runs were added after the ramp at 100 users the same day, where 119 approved runs waited for a resume that the admission limit had answered 503.

`cleanup`

async def cleanup(): ...

Clean up resources

`cleanup_mcp_servers`

async def cleanup_mcp_servers(): ...

Clean up MCP servers without removing the agent and the config

`clear_session_history`

async def clear_session_history(session_id: Optional[str] = None): ...

Clear session history for a specific session ID or all history

`connect_mcp_servers`

async def connect_mcp_servers(): ...

Connect to MCP servers if MCP tools are configured

`deny_budget`

async def deny_budget(run_id: str, *, approver: str, note: Optional[str] = None, request_id: Optional[str] = None) -> Dict[str, Any]: ...

Refuse a waiting run’s budget: resume(run_id) ends it cleanly.

`export_trace`

async def export_trace(identifier: str | None = None, *, session_id: str | None = None, trace_id: str | None = None, run_id: str | None = None, exporters: Optional[List[Any]] = None, normalize: bool = True, strict: bool = False) -> list[Dict[str, Any]]: ...

Export a telemetry trace through configured or supplied exporters.

`generate_run_id`

def generate_run_id() -> str: ...

Generate a unique run ID inside a session.

`generate_session_id`

def generate_session_id() -> str: ...

Generate a new session ID for the session

`get_latest_trace`

async def get_latest_trace(session_id: str, *, normalize: bool = False) -> Dict[str, Any] | None: ...

This agent’s latest trace in a session. A sub-agent shares its lead’s session; its traces are the lead’s trace family, not this.

`get_memory_store_type`

async def get_memory_store_type() -> str: ...

Get the current memory store type.

`get_metrics`

async def get_metrics() -> Dict[str, Any]: ...

Get the cumulative metrics for the lifecycle of the agent.

`get_run`

async def get_run(run_id: str) -> Optional[Dict[str, Any]]: ...

A run’s durable record, or None if it has none. The record is kept in the agent’s memory store: status, step, usage, trace IDs, each tool call’s state (arguments as a digest), and each approval with the arguments of the call it is for.

`get_run_trajectory`

async def get_run_trajectory(run_id: str) -> Dict[str, Any] | None: ...

One durable run as a single story across its trace segments (each pause, resume, recovery, or new attempt is a segment), with totals summed over the segments. The run’s saved conversation is not included. A segment whose trace is no longer stored (traces are pruned after telemetry_config["retention_days"], run records after run_retention_days) has trace_kept: False; traces_missing counts them, and totals cover only the traces still kept, so they are empty rather than zero when none is. The record’s usage remains.

`get_session_history`

async def get_session_history(session_id: str) -> List[Dict[str, Any]]: ...

Get session history for a specific session ID

`get_telemetry_events_after`

async def get_telemetry_events_after(*, cursor: str | None, trace_id: str | None = None, run_id: str | None = None, session_id: str | None = None, task_id: str | None = None, event_types: tuple[str, ...] | None = None): ...

The telemetry events stored after cursor, as a list, in the order they were recorded, filtered like stream_telemetry_after; it returns at once instead of waiting for more. cursor=None returns every stored event in scope. Poll with the last event’s stream_cursor to get only what is new.

`get_telemetry_stream_cursor`

async def get_telemetry_stream_cursor(*, trace_id: str | None = None, run_id: str | None = None, session_id: str | None = None, task_id: str | None = None, event_types: tuple[str, ...] | None = None) -> str | None: ...

The store’s position now: a cursor to pass to stream_telemetry_after or get_telemetry_events_after to get only events recorded from here on. Take it before starting a run to follow that run from its first event. Cursors are opaque strings that increase with every event.

`get_telemetry_trace`

async def get_telemetry_trace(trace_id: str, *, normalize: bool = False) -> Dict[str, Any] | None: ...

The raw trace of one trace segment, as a dict: its spans and events as recorded, or None if it is not stored (never recorded, or pruned after telemetry_config["retention_days"]). normalize=True sorts spans by start and events by sequence, and marks capture gaps and missing references. For a readable story use get_trajectory (one segment) or get_run_trajectory (a whole run, across pauses and resumes).

`get_trace`

async def get_trace(identifier: str | None = None, *, session_id: str | None = None, trace_id: str | None = None, run_id: str | None = None, normalize: bool = False) -> Dict[str, Any] | None: ...

Return telemetry trace data.

`get_trace_family`

async def get_trace_family(identifier: str | None = None, *, trace_id: str | None = None, run_id: str | None = None, normalize: bool = False) -> list[Dict[str, Any]]: ...

Return all locally stored traces linked to one execution boundary. A family follows explicit parent links in both directions. For a run_id seed, every trace carrying that run id is included before linked ancestors and descendants are added. This avoids treating the latest trace for a session as the complete execution history.

`get_trajectory`

async def get_trajectory(identifier: str | None = None, *, trace_id: str | None = None, run_id: str | None = None, include_children: bool = True, max_depth: int = 5) -> Dict[str, Any] | None: ...

One trace segment as an ordered trajectory: steps, model calls, tool calls and their observations, from request to answer or pause. A run that paused and resumed has one segment per stretch of work; get_run_trajectory(run_id) gives the whole run. Look up by trace_id (exact) or run_id (the latest segment of that run; the others are listed in other_trace_ids_for_run). Delegated child runs are nested under the tool call that started them. See build_trajectory for the shape.

`grant_budget`

async def grant_budget(run_id: str, *, amount: Optional[float] = None, approver: str, note: Optional[str] = None, request_id: Optional[str] = None) -> Dict[str, Any]: ...

Add to the budget a waiting run ran out of, so it can carry on. The policy is not changed: this is a recorded exception to one budget, with the name of whoever made it. amount defaults to what the run said it was short of. resume(run_id) then continues the work.

`initialize`

async def initialize(): ...

Initialize the agent resources (memory, config, tools)

`interrupt`

async def interrupt(run_id: str) -> Dict[str, Any]: ...

Ask a running run to stop at its next step boundary; it becomes interrupted and resume continues it.

`list_all_available_tools`

async def list_all_available_tools(): ...

Every tool the model is offered: MCP, local and built-in tools, and a delegate_<name> tool for each of sub_agents.

`list_runs`

async def list_runs(session_id: Optional[str] = None, status: Optional[str] = None, limit: int = 100) -> List[Dict[str, Any]]: ...

Run records, oldest first, optionally for one session or status.

`list_telemetry_traces`

async def list_telemetry_traces(trace_filter: TraceFilter | None = None, *, trace_id: str | None = None, run_id: str | None = None, session_id: str | None = None, task_id: str | None = None, agent_id: str | None = None, workflow_id: str | None = None, model: str | None = None, status: TraceStatus | str | None = None, normalize: bool = False) -> list[Dict[str, Any]]: ...

Stored traces matching every filter given, oldest first (by start time, then trace ID), as dicts. Filter with keywords or one TraceFilter, not both. A run that paused and resumed has one trace per segment; run_id= lists them all. normalize is as in get_telemetry_trace.

`prune_runs`

async def prune_runs() -> Dict[str, Any]: ...

Remove finished run records older than run_retention_days now (30 by default; None keeps every record). A run still waiting for a person or a resume is never removed. Returns what was removed.

`prune_telemetry`

async def prune_telemetry() -> Dict[str, Any]: ...

Apply the configured trace and payload retention now. Expired finished traces are removed first; payloads are then pruned by their own retention window, except any payload still referenced by a kept trace. Returns what was removed.

`prune_telemetry_payloads`

async def prune_telemetry_payloads(retention_days: int | None = None) -> int: ...

Prune payloads while retaining references in currently stored traces.

`read_telemetry_payload`

async def read_telemetry_payload(reference: str) -> Any: ...

Read a redacted oversized payload referenced by telemetry.

`record_outcome`

async def record_outcome(run_id: str, *, source: str, reward: float | None = None, label: str | None = None, detail: Dict[str, Any] | None = None) -> Dict[str, Any]: ...

Record what a run turned out to be worth, whenever that is known. A run’s own result is not its outcome: a pull request is merged an hour later, a customer accepts an answer the next day, CI runs the tests afterwards. source says who is reporting it (github, reviewer, an evaluator’s name), reward is the number a trainer or an evaluator uses, label is what it is called, and detail is anything else worth keeping. A run may gather several. The outcome goes on the run’s record and into its trace.

`release_claim`

async def release_claim(claim: Dict[str, Any]) -> None: ...

Stop keeping a claim’s lease alive (a claim nobody will resume, or one the run has taken over). Safe to call more than once.

`resolve_approval`

async def resolve_approval(run_id: str, approval_id: str, *, decision: str, approver: str, note: Optional[str] = None, arguments: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: ...

Decide an approval a run is waiting for. decision is "approve" or "deny". A denial’s note reaches the model. arguments approves an edited call instead of the one asked for. The decision applies once, to that exact request.

`resume`

async def resume(run_id: str, on_event: Any = None, *, trigger: str = 'explicit') -> Dict[str, Any]: ...

Continue a run: one waiting for approval once every approval is decided (see resolve_approval), or one whose process stopped (its heartbeat is older than run_lease_seconds). Completed tool calls never run again. trigger says who resumed it, for the run’s trace and the omniserve_runs_resumed_total metric: a person or client (explicit), or the background supervisor retrying (background_retry).

`resume_claimed`

async def resume_claimed(claim: Dict[str, Any], on_event: Any = None) -> Dict[str, Any]: ...

Continue a run claim_orphaned_runs returned, with the same durable rules as resume: completed calls never run again, and a call that did not finish becomes unknown_outcome unless its tool is idempotent.

`run`

async def run(query: str, session_id: Optional[str] = None, run_id: Optional[str] = None, on_event: Any = None, *, tags: Optional[List[str]] = None, provenance: Optional[Dict[str, Any]] = None, _resume: Optional[Dict[str, Any]] = None, _resume_cause: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: ...

Run the agent with a query and optional session ID.

`steer`

async def steer(run_id: str, message: str, *, sender: Optional[str] = None) -> Dict[str, Any]: ...

Send a message to a run; it arrives as a user message at the run’s next step boundary (or when a waiting or interrupted run resumes). The message is user input: the injection guardrail checks it first, and a blocked message is never queued.

`stream`

def stream(query: str, session_id: str | None = None, run_id: str | None = None): ...

Stream live intermediate text and one terminal result from the same run loop. Use contextlib.aclosing when stopping iteration early. Deltas are not replayed from telemetry; the completed answer is persisted normally.

`stream_telemetry_after`

def stream_telemetry_after(*, cursor: str | None, trace_id: str | None = None, run_id: str | None = None, session_id: str | None = None, task_id: str | None = None, event_types: tuple[str, ...] | None = None): ...

Yield telemetry events after cursor, then keep yielding new ones as they are recorded; it does not end on its own. Events come in the order they were recorded, each once, filtered to the trace, run, session or task and the event_types given. cursor=None starts from the first stored event. Each event carries its stream_cursor: keep the last one to resume after a disconnect. A reader that falls 1,000 events behind is stopped with an error rather than silently skipping events.

`switch_memory_store`

async def switch_memory_store(memory_store_type: str): ...

Switch to a different memory store type.

`telemetry_retention_status`

def telemetry_retention_status() -> Dict[str, Any]: ...

Report the retention policy and the most recent cleanup results: traces, payloads, and run records (runs).

`training_records`

async def training_records(*, run_id: str | None = None, session_id: str | None = None, trace_ids: List[str] | None = None, limit: int | None = None) -> List[Dict[str, Any]]: ...

Finished runs as one record each, for a trainer or an evaluator. Each record holds what the model was sent at every step, what it produced (its token details when they were recorded), what the tools answered, the policy that served it, the run’s totals, and the outcomes attached to it. A run that paused and resumed is one record with the steps of every trace segment. A run still running or waiting, or one recorded without model prompts, is left out.

​OmniCoreAgent

​Constructor

​Methods

​abandon_run

​budget_status

​claim_orphaned_runs

​cleanup

​cleanup_mcp_servers

​clear_session_history

​connect_mcp_servers

​deny_budget

​export_trace

​generate_run_id

​generate_session_id

​get_latest_trace

​get_memory_store_type

​get_metrics

​get_run

​get_run_trajectory

​get_session_history

​get_telemetry_events_after

​get_telemetry_stream_cursor

​get_telemetry_trace

​get_trace

​get_trace_family

​get_trajectory

​grant_budget

​initialize

​interrupt

​list_all_available_tools

​list_runs

​list_telemetry_traces

​prune_runs

​prune_telemetry

​prune_telemetry_payloads

​read_telemetry_payload

​record_outcome

​release_claim

​resolve_approval

​resume

​resume_claimed

​run

​steer

​stream

​stream_telemetry_after

​switch_memory_store

​telemetry_retention_status

​training_records