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`, `base_url` and others (see Models). The key comes from `LLM_API_KEY`.
`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`, or a list of functions.
`sub_agents`	`None`	Other `OmniCoreAgent` instances this one may hand a task to, by name, through a `call_sub_agent` 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.

`budget_status`

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

Every budget covering a run, with its limit and what it has spent.

`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: ...

No description in the code yet.

`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.

`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.

`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): ...

No description in the code yet.

`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: ...

No description in the code yet.

`get_telemetry_trace`

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

No description in the code yet.

`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.

`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: ...

Return one run as an ordered trajectory, from request to final answer.

`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.

`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(): ...

List all available tools (MCP and local)

`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]]: ...

No description in the code yet.

`prune_telemetry`

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

Apply the configured trace and payload retention now.

`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.

`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.

`resume`

async def resume(run_id: str, on_event: Any = None) -> 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.

`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) -> 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).

`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.

`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): ...

No description in the code yet.

`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.

`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.

​OmniCoreAgent

​Constructor

​Methods

​abandon_run

​budget_status

​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_telemetry

​prune_telemetry_payloads

​read_telemetry_payload

​record_outcome

​resolve_approval

​resume

​run

​steer

​stream

​stream_telemetry_after

​switch_memory_store

​telemetry_retention_status

​training_records

OmniCoreAgent

Constructor

Methods

`abandon_run`

`budget_status`

`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_telemetry`

`prune_telemetry_payloads`

`read_telemetry_payload`

`record_outcome`

`resolve_approval`

`resume`

`run`

`steer`

`stream`

`stream_telemetry_after`

`switch_memory_store`

`telemetry_retention_status`

`training_records`