> ## Documentation Index > Fetch the complete documentation index at: https://docs-omnicoreagent.omnirexfloralabs.com/llms.txt > Use this file to discover all available pages before exploring further. # OmniCoreAgent > The agent: its constructor and every public method # OmniCoreAgent ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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](/docs/reference/agent-config), and `telemetry_config` the [telemetry settings](/docs/reference/telemetry-config). ## Methods ### `abandon_run` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} async def cleanup(): ... ``` Clean up resources ### `cleanup_mcp_servers` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} async def cleanup_mcp_servers(): ... ``` Clean up MCP servers without removing the agent and the config ### `clear_session_history` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} async def clear_session_history(session_id: Optional[str] = None): ... ``` Clear session history for a specific session ID or all history ### `connect_mcp_servers` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} async def connect_mcp_servers(): ... ``` Connect to MCP servers if MCP tools are configured ### `deny_budget` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} def generate_run_id() -> str: ... ``` Generate a unique run ID inside a session. ### `generate_session_id` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} def generate_session_id() -> str: ... ``` Generate a new session ID for the session ### `get_latest_trace` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} async def get_memory_store_type() -> str: ... ``` Get the current memory store type. ### `get_metrics` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} async def get_metrics() -> Dict[str, Any]: ... ``` Get the cumulative metrics for the lifecycle of the agent. ### `get_run` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} async def get_session_history(session_id: str) -> List[Dict[str, Any]]: ... ``` Get session history for a specific session ID ### `get_telemetry_events_after` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} async def get_telemetry_trace(trace_id: str, *, normalize: bool = False) -> Dict[str, Any] | None: ... ``` *No description in the code yet.* ### `get_trace` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} async def initialize(): ... ``` Initialize the agent resources (memory, config, tools) ### `interrupt` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} async def list_all_available_tools(): ... ``` List all available tools (MCP and local) ### `list_runs` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} async def prune_telemetry() -> Dict[str, Any]: ... ``` Apply the configured trace and payload retention now. ### `prune_telemetry_payloads` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} async def prune_telemetry_payloads(retention_days: int | None = None) -> int: ... ``` Prune payloads while retaining references in currently stored traces. ### `read_telemetry_payload` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} async def read_telemetry_payload(reference: str) -> Any: ... ``` Read a redacted oversized payload referenced by telemetry. ### `record_outcome` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} async def switch_memory_store(memory_store_type: str): ... ``` Switch to a different memory store type. ### `telemetry_retention_status` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} def telemetry_retention_status() -> Dict[str, Any]: ... ``` Report the retention policy and the most recent cleanup results. ### `training_records` ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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.