Utility API reference

General utilities

Checkpointer

Source code in src/ursa/util/__init__.py

class Checkpointer:
    @classmethod
    def from_workspace(
        cls,
        workspace: Path,
        db_dir: str = "db",
        db_name: str = "checkpointer.db",
    ) -> SqliteSaver:
        (db_path := workspace / db_dir).mkdir(parents=True, exist_ok=True)
        conn = sqlite3.connect(str(db_path / db_name), check_same_thread=False)
        return SqliteSaver(conn)

    @classmethod
    async def async_from_workspace(
        cls,
        workspace: Path,
        db_dir: str = "db",
        db_name: str = "checkpointer.db",
    ) -> AsyncSqliteSaver:
        """Make an async SQLite checkpointer under a workspace directory."""
        (db_path := workspace / db_dir).mkdir(parents=True, exist_ok=True)
        conn = await aiosqlite.connect(str(db_path / db_name))
        return AsyncSqliteSaver(conn)

    @classmethod
    def from_path(
        cls, db_path: Path, db_name: str = "checkpointer.db"
    ) -> SqliteSaver:
        """Make checkpointer sqlite db.

        Args
        ====
        * db_path: The path to the SQLite database file (e.g. ./checkpoint.db) to be created.
        """

        db_path.parent.mkdir(parents=True, exist_ok=True)
        conn = sqlite3.connect(str(db_path / db_name), check_same_thread=False)
        return SqliteSaver(conn)

    @classmethod
    async def async_from_path(
        cls, db_path: Path, db_name: str = "checkpointer.db"
    ) -> AsyncSqliteSaver:
        """Make an async SQLite checkpointer under a database path."""
        db_path.parent.mkdir(parents=True, exist_ok=True)
        conn = await aiosqlite.connect(str(db_path / db_name))
        return AsyncSqliteSaver(conn)

async_from_path(db_path, db_name='checkpointer.db') async classmethod

Make an async SQLite checkpointer under a database path.

Source code in src/ursa/util/__init__.py

@classmethod
async def async_from_path(
    cls, db_path: Path, db_name: str = "checkpointer.db"
) -> AsyncSqliteSaver:
    """Make an async SQLite checkpointer under a database path."""
    db_path.parent.mkdir(parents=True, exist_ok=True)
    conn = await aiosqlite.connect(str(db_path / db_name))
    return AsyncSqliteSaver(conn)

async_from_workspace(workspace, db_dir='db', db_name='checkpointer.db') async classmethod

Make an async SQLite checkpointer under a workspace directory.

Source code in src/ursa/util/__init__.py

@classmethod
async def async_from_workspace(
    cls,
    workspace: Path,
    db_dir: str = "db",
    db_name: str = "checkpointer.db",
) -> AsyncSqliteSaver:
    """Make an async SQLite checkpointer under a workspace directory."""
    (db_path := workspace / db_dir).mkdir(parents=True, exist_ok=True)
    conn = await aiosqlite.connect(str(db_path / db_name))
    return AsyncSqliteSaver(conn)

from_path(db_path, db_name='checkpointer.db') classmethod

Make checkpointer sqlite db.

Args

• db_path: The path to the SQLite database file (e.g. ./checkpoint.db) to be created.

Source code in src/ursa/util/__init__.py

@classmethod
def from_path(
    cls, db_path: Path, db_name: str = "checkpointer.db"
) -> SqliteSaver:
    """Make checkpointer sqlite db.

    Args
    ====
    * db_path: The path to the SQLite database file (e.g. ./checkpoint.db) to be created.
    """

    db_path.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(db_path / db_name), check_same_thread=False)
    return SqliteSaver(conn)

Event logging

Event helpers

ursa.util.events.DEFAULT_EVENT_NAME = 'ursa_agent_progress' module-attribute

Name used for every structured URSA progress event.

ursa.util.events.ProgressEvents dataclass

Emit standardized URSA progress events for a single source.

Parameters

name: Stable short identifier included in every payload. config: Runnable config propagated by LangGraph/LangChain. When None, all methods become safe no-ops. event_name: Underlying LangChain custom event name. The default preserves the contract already used by the deployed branch work. name_key: Payload key used for the source identifier, for example "agent" or "tool". default_payload: Extra fields merged into every emitted payload.

Source code in src/ursa/util/events.py

@dataclass(slots=True)
class ProgressEvents:
    """Emit standardized URSA progress events for a single source.

    Parameters
    ----------
    name:
        Stable short identifier included in every payload.
    config:
        Runnable config propagated by LangGraph/LangChain. When ``None``, all
        methods become safe no-ops.
    event_name:
        Underlying LangChain custom event name. The default preserves the
        contract already used by the deployed branch work.
    name_key:
        Payload key used for the source identifier, for example ``"agent"`` or
        ``"tool"``.
    default_payload:
        Extra fields merged into every emitted payload.
    """

    name: str
    config: RunnableConfig | None = None
    event_name: str = DEFAULT_EVENT_NAME
    name_key: str = "name"
    default_payload: dict[str, Any] = field(default_factory=dict)

    def emit(
        self,
        message: str,
        *,
        stage: str,
        **payload: Any,
    ) -> dict[str, Any] | None:
        """Emit one synchronous progress event.

        Parameters
        ----------
        message:
            Concise human-readable description of the progress update.
        stage:
            Stable machine-readable stage name.
        **payload:
            Additional serializable fields merged into the event payload.

        Returns
        -------
        dict or None
            The dispatched payload, or ``None`` when no runnable config was
            supplied and event emission is disabled.
        """
        if self.config is None:
            return None
        body = self._payload(message=message, stage=stage, **payload)
        dispatch_custom_event(self.event_name, body, config=self.config)
        return body

    async def aemit(
        self,
        message: str,
        *,
        stage: str,
        **payload: Any,
    ) -> dict[str, Any] | None:
        """Asynchronously emit one progress event.

        Parameters and return behavior match ``emit``. Dispatch waits for
        LangChain's asynchronous custom-event manager to complete.
        """
        if self.config is None:
            return None
        body = self._payload(message=message, stage=stage, **payload)
        await adispatch_custom_event(
            self.event_name,
            body,
            config=self.config,
        )
        return body

    def range(
        self,
        stage: str,
        start: str,
        *,
        done: str | None = None,
        error: str | None = None,
        **payload: Any,
    ) -> EventRange:
        """Create a context manager for an operation lifecycle.

        Parameters
        ----------
        stage:
            Stable stage included in every range event.
        start:
            Message emitted with ``phase="start"`` on entry.
        done:
            Message emitted with ``phase="end"`` after successful completion.
            Defaults to ``start``.
        error:
            Message emitted with ``phase="error"`` when an exception escapes.
            Defaults to ``start``.
        **payload:
            Fields included in the start event and initial terminal payload.

        Returns
        -------
        EventRange
            A synchronous and asynchronous context manager. Call
            ``EventRange.update`` inside the block to add terminal-only
            fields such as results or artifacts.
        """
        return EventRange(
            events=self,
            stage=stage,
            start_message=start,
            done_message=done,
            error_message=error,
            payload=dict(payload),
        )

    def _payload(
        self,
        *,
        message: str,
        stage: str,
        **payload: Any,
    ) -> dict[str, Any]:
        return {
            self.name_key: self.name,
            "stage": stage,
            "message": message,
            "monotonic_timestamp_ns": monotonic_ns(),
            **self.default_payload,
            **payload,
        }

aemit(message, *, stage, **payload) async

Asynchronously emit one progress event.

Parameters and return behavior match emit. Dispatch waits for LangChain's asynchronous custom-event manager to complete.

Source code in src/ursa/util/events.py

async def aemit(
    self,
    message: str,
    *,
    stage: str,
    **payload: Any,
) -> dict[str, Any] | None:
    """Asynchronously emit one progress event.

    Parameters and return behavior match ``emit``. Dispatch waits for
    LangChain's asynchronous custom-event manager to complete.
    """
    if self.config is None:
        return None
    body = self._payload(message=message, stage=stage, **payload)
    await adispatch_custom_event(
        self.event_name,
        body,
        config=self.config,
    )
    return body

emit(message, *, stage, **payload)

Emit one synchronous progress event.

Parameters

message: Concise human-readable description of the progress update. stage: Stable machine-readable stage name. **payload: Additional serializable fields merged into the event payload.

Returns

dict or None The dispatched payload, or None when no runnable config was supplied and event emission is disabled.

Source code in src/ursa/util/events.py

def emit(
    self,
    message: str,
    *,
    stage: str,
    **payload: Any,
) -> dict[str, Any] | None:
    """Emit one synchronous progress event.

    Parameters
    ----------
    message:
        Concise human-readable description of the progress update.
    stage:
        Stable machine-readable stage name.
    **payload:
        Additional serializable fields merged into the event payload.

    Returns
    -------
    dict or None
        The dispatched payload, or ``None`` when no runnable config was
        supplied and event emission is disabled.
    """
    if self.config is None:
        return None
    body = self._payload(message=message, stage=stage, **payload)
    dispatch_custom_event(self.event_name, body, config=self.config)
    return body

range(stage, start, *, done=None, error=None, **payload)

Create a context manager for an operation lifecycle.

Parameters

stage: Stable stage included in every range event. start: Message emitted with phase="start" on entry. done: Message emitted with phase="end" after successful completion. Defaults to start. error: Message emitted with phase="error" when an exception escapes. Defaults to start. **payload: Fields included in the start event and initial terminal payload.

Returns

EventRange A synchronous and asynchronous context manager. Call EventRange.update inside the block to add terminal-only fields such as results or artifacts.

Source code in src/ursa/util/events.py

def range(
    self,
    stage: str,
    start: str,
    *,
    done: str | None = None,
    error: str | None = None,
    **payload: Any,
) -> EventRange:
    """Create a context manager for an operation lifecycle.

    Parameters
    ----------
    stage:
        Stable stage included in every range event.
    start:
        Message emitted with ``phase="start"`` on entry.
    done:
        Message emitted with ``phase="end"`` after successful completion.
        Defaults to ``start``.
    error:
        Message emitted with ``phase="error"`` when an exception escapes.
        Defaults to ``start``.
    **payload:
        Fields included in the start event and initial terminal payload.

    Returns
    -------
    EventRange
        A synchronous and asynchronous context manager. Call
        ``EventRange.update`` inside the block to add terminal-only
        fields such as results or artifacts.
    """
    return EventRange(
        events=self,
        stage=stage,
        start_message=start,
        done_message=done,
        error_message=error,
        payload=dict(payload),
    )

ursa.util.events.AgentEvents

Bases: ProgressEvents

Emit standardized progress events for one agent.

Parameters

agent: Stable agent name stored in the agent payload field. config: Active LangChain runnable configuration. Without a config, emission methods are safe no-ops. event_name: Custom event channel, normally DEFAULT_EVENT_NAME.

Source code in src/ursa/util/events.py

class AgentEvents(ProgressEvents):
    """Emit standardized progress events for one agent.

    Parameters
    ----------
    agent:
        Stable agent name stored in the ``agent`` payload field.
    config:
        Active LangChain runnable configuration. Without a config, emission
        methods are safe no-ops.
    event_name:
        Custom event channel, normally ``DEFAULT_EVENT_NAME``.
    """

    def __init__(
        self,
        agent: str,
        config: RunnableConfig | None = None,
        event_name: str = DEFAULT_EVENT_NAME,
    ) -> None:
        super().__init__(
            name=agent,
            config=config,
            event_name=event_name,
            name_key="agent",
        )

ursa.util.events.EnvironmentEvents

Bases: ProgressEvents

Emit standardized URSA progress events for an environment.

Environment events use the same custom LangChain event channel as agent and tool events so a single callback recorder can capture complete nested runs. Environments may also be invoked directly rather than from inside a LangChain runnable. In that case LangChain refuses ad-hoc custom events because there is no parent run ID, so this helper falls back to directly notifying callbacks from the runnable config.

Parameters

environment: Stable environment name stored in the environment payload field. config: Active runnable configuration and callback collection. event_name: Custom event channel, normally DEFAULT_EVENT_NAME. environment_type: Optional environment kind added to every payload. environment_id: Optional stable environment identifier added to every payload. path: Optional hierarchical environment path added to every payload.

Source code in src/ursa/util/events.py

class EnvironmentEvents(ProgressEvents):
    """Emit standardized URSA progress events for an environment.

    Environment events use the same custom LangChain event channel as agent and
    tool events so a single callback recorder can capture complete nested runs.
    Environments may also be invoked directly rather than from inside a
    LangChain runnable. In that case LangChain refuses ad-hoc custom events
    because there is no parent run ID, so this helper falls back to directly
    notifying callbacks from the runnable config.

    Parameters
    ----------
    environment:
        Stable environment name stored in the ``environment`` payload field.
    config:
        Active runnable configuration and callback collection.
    event_name:
        Custom event channel, normally ``DEFAULT_EVENT_NAME``.
    environment_type:
        Optional environment kind added to every payload.
    environment_id:
        Optional stable environment identifier added to every payload.
    path:
        Optional hierarchical environment path added to every payload.
    """

    def __init__(
        self,
        environment: str,
        config: RunnableConfig | None = None,
        event_name: str = DEFAULT_EVENT_NAME,
        *,
        environment_type: str | None = None,
        environment_id: str | None = None,
        path: list[str] | None = None,
    ) -> None:
        default_payload: dict[str, Any] = {}
        if environment_type is not None:
            default_payload["environment_type"] = environment_type
        if environment_id is not None:
            default_payload["environment_id"] = environment_id
        if path is not None:
            default_payload["path"] = path
        super().__init__(
            name=environment,
            config=config,
            event_name=event_name,
            name_key="environment",
            default_payload=default_payload,
        )

    def emit(
        self,
        message: str,
        *,
        stage: str,
        **payload: Any,
    ) -> dict[str, Any] | None:
        """Emit an environment event synchronously.

        When LangChain reports that no parent run exists, the event is sent
        directly to callbacks in ``config``. This supports environments that
        are invoked outside a runnable while preserving the normal custom-event
        path for nested runs.
        """
        if self.config is None:
            return None
        body = self._payload(message=message, stage=stage, **payload)
        try:
            dispatch_custom_event(self.event_name, body, config=self.config)
        except RuntimeError as exc:
            if not self._is_missing_parent_run_error(exc):
                raise
            self._direct_emit(body)
        return body

    async def aemit(
        self,
        message: str,
        *,
        stage: str,
        **payload: Any,
    ) -> dict[str, Any] | None:
        """Asynchronously emit an environment event.

        This method provides the same direct-callback fallback as ``emit`` and
        awaits asynchronous callback implementations when necessary.
        """
        if self.config is None:
            return None
        body = self._payload(message=message, stage=stage, **payload)
        try:
            await adispatch_custom_event(
                self.event_name,
                body,
                config=self.config,
            )
        except RuntimeError as exc:
            if not self._is_missing_parent_run_error(exc):
                raise
            await self._adirect_emit(body)
        return body

    @staticmethod
    def _is_missing_parent_run_error(exc: RuntimeError) -> bool:
        return (
            "Unable to dispatch an adhoc event without a parent run id"
            in str(exc)
        )

    def _callbacks(self) -> list[Any]:
        if self.config is None:
            return []
        callbacks = self.config.get("callbacks", [])
        if callbacks is None:
            return []
        if isinstance(callbacks, list):
            return callbacks
        return [callbacks]

    def _direct_emit(self, body: dict[str, Any]) -> None:
        run_id = uuid4()
        for callback in self._callbacks():
            handler = getattr(callback, "on_custom_event", None)
            if handler is None:
                continue
            handler(self.event_name, body, run_id=run_id)

    async def _adirect_emit(self, body: dict[str, Any]) -> None:
        run_id = uuid4()
        for callback in self._callbacks():
            handler = getattr(callback, "on_custom_event", None)
            if handler is None:
                continue
            result = handler(self.event_name, body, run_id=run_id)
            if inspect.isawaitable(result):
                await result

aemit(message, *, stage, **payload) async

Asynchronously emit an environment event.

This method provides the same direct-callback fallback as emit and awaits asynchronous callback implementations when necessary.

Source code in src/ursa/util/events.py

async def aemit(
    self,
    message: str,
    *,
    stage: str,
    **payload: Any,
) -> dict[str, Any] | None:
    """Asynchronously emit an environment event.

    This method provides the same direct-callback fallback as ``emit`` and
    awaits asynchronous callback implementations when necessary.
    """
    if self.config is None:
        return None
    body = self._payload(message=message, stage=stage, **payload)
    try:
        await adispatch_custom_event(
            self.event_name,
            body,
            config=self.config,
        )
    except RuntimeError as exc:
        if not self._is_missing_parent_run_error(exc):
            raise
        await self._adirect_emit(body)
    return body

emit(message, *, stage, **payload)

Emit an environment event synchronously.

When LangChain reports that no parent run exists, the event is sent directly to callbacks in config. This supports environments that are invoked outside a runnable while preserving the normal custom-event path for nested runs.

Source code in src/ursa/util/events.py

def emit(
    self,
    message: str,
    *,
    stage: str,
    **payload: Any,
) -> dict[str, Any] | None:
    """Emit an environment event synchronously.

    When LangChain reports that no parent run exists, the event is sent
    directly to callbacks in ``config``. This supports environments that
    are invoked outside a runnable while preserving the normal custom-event
    path for nested runs.
    """
    if self.config is None:
        return None
    body = self._payload(message=message, stage=stage, **payload)
    try:
        dispatch_custom_event(self.event_name, body, config=self.config)
    except RuntimeError as exc:
        if not self._is_missing_parent_run_error(exc):
            raise
        self._direct_emit(body)
    return body

ursa.util.events.ToolEvents

Bases: ProgressEvents

Emit standardized progress events for one tool invocation.

Parameters

tool: Stable tool name stored in the tool payload field. config: Active runnable configuration. Without a config, emission methods are safe no-ops. event_name: Custom event channel, normally DEFAULT_EVENT_NAME. tool_call_id: Optional LangGraph tool-call identifier added to every payload. owner_payload: Optional agent or environment identity fields added to every payload.

Source code in src/ursa/util/events.py

class ToolEvents(ProgressEvents):
    """Emit standardized progress events for one tool invocation.

    Parameters
    ----------
    tool:
        Stable tool name stored in the ``tool`` payload field.
    config:
        Active runnable configuration. Without a config, emission methods are
        safe no-ops.
    event_name:
        Custom event channel, normally ``DEFAULT_EVENT_NAME``.
    tool_call_id:
        Optional LangGraph tool-call identifier added to every payload.
    owner_payload:
        Optional agent or environment identity fields added to every payload.
    """

    def __init__(
        self,
        tool: str,
        config: RunnableConfig | None = None,
        event_name: str = DEFAULT_EVENT_NAME,
        *,
        tool_call_id: str | None = None,
        owner_payload: Mapping[str, Any] | None = None,
    ) -> None:
        default_payload: dict[str, Any] = {}
        if owner_payload:
            default_payload.update(owner_payload)
        if tool_call_id is not None:
            default_payload["tool_call_id"] = tool_call_id
        super().__init__(
            name=tool,
            config=config,
            event_name=event_name,
            name_key="tool",
            default_payload=default_payload,
        )

    @staticmethod
    def _owner_payload_from_runtime(
        runtime: ToolRuntime[Any],
    ) -> dict[str, Any]:
        """Return agent/member identity fields recorded on a tool runtime."""
        config = runtime.config if isinstance(runtime.config, Mapping) else {}
        metadata = config.get("metadata")
        if not isinstance(metadata, Mapping):
            metadata = {}
        context = getattr(runtime, "context", None)

        environment_id = metadata.get("environment_id")
        member = metadata.get("environment_member")
        member_id = metadata.get("environment_member_id")
        member_path = metadata.get("environment_member_path")
        member_role = metadata.get("environment_member_role")

        agent = metadata.get("agent") or member
        agent_id = metadata.get("agent_id") or member_id
        if agent is None and context is not None:
            agent = getattr(context, "agent_name", None)
        if agent_id is None and environment_id and agent:
            agent_id = f"{environment_id}.{agent}"

        payload: dict[str, Any] = {}
        if agent is not None:
            payload["agent"] = str(agent)
        if agent_id is not None:
            payload["agent_id"] = str(agent_id)
        if member is not None:
            payload["environment_member"] = str(member)
        if member_id is not None:
            payload["environment_member_id"] = str(member_id)
        if member_role is not None:
            payload["environment_member_role"] = str(member_role)
        if environment_id is not None:
            payload["environment_id"] = str(environment_id)
        if member_path is not None:
            payload["environment_member_path"] = member_path
        return payload

    @classmethod
    def from_runtime(
        cls,
        tool: str,
        runtime: ToolRuntime[Any] | None,
        *,
        event_name: str = DEFAULT_EVENT_NAME,
    ) -> ToolEvents:
        """Create a tool event helper from a LangGraph runtime.

        The runtime supplies the runnable config, tool-call ID, and owning
        agent or environment metadata. Passing ``None`` returns a helper with
        emission disabled, which is useful for optional runtime integrations.

        Parameters
        ----------
        tool:
            Stable tool name stored in emitted payloads.
        runtime:
            Current tool runtime, or ``None`` when no runtime is available.
        event_name:
            Custom event channel, normally ``DEFAULT_EVENT_NAME``.

        Returns
        -------
        ToolEvents
            A helper configured from the runtime metadata.
        """
        if runtime is None:
            return cls(tool=tool, event_name=event_name)
        return cls(
            tool=tool,
            config=runtime.config,
            event_name=event_name,
            tool_call_id=runtime.tool_call_id,
            owner_payload=cls._owner_payload_from_runtime(runtime),
        )

from_runtime(tool, runtime, *, event_name=DEFAULT_EVENT_NAME) classmethod

Create a tool event helper from a LangGraph runtime.

The runtime supplies the runnable config, tool-call ID, and owning agent or environment metadata. Passing None returns a helper with emission disabled, which is useful for optional runtime integrations.

Parameters

tool: Stable tool name stored in emitted payloads. runtime: Current tool runtime, or None when no runtime is available. event_name: Custom event channel, normally DEFAULT_EVENT_NAME.

Returns

ToolEvents A helper configured from the runtime metadata.

Source code in src/ursa/util/events.py

@classmethod
def from_runtime(
    cls,
    tool: str,
    runtime: ToolRuntime[Any] | None,
    *,
    event_name: str = DEFAULT_EVENT_NAME,
) -> ToolEvents:
    """Create a tool event helper from a LangGraph runtime.

    The runtime supplies the runnable config, tool-call ID, and owning
    agent or environment metadata. Passing ``None`` returns a helper with
    emission disabled, which is useful for optional runtime integrations.

    Parameters
    ----------
    tool:
        Stable tool name stored in emitted payloads.
    runtime:
        Current tool runtime, or ``None`` when no runtime is available.
    event_name:
        Custom event channel, normally ``DEFAULT_EVENT_NAME``.

    Returns
    -------
    ToolEvents
        A helper configured from the runtime metadata.
    """
    if runtime is None:
        return cls(tool=tool, event_name=event_name)
    return cls(
        tool=tool,
        config=runtime.config,
        event_name=event_name,
        tool_call_id=runtime.tool_call_id,
        owner_payload=cls._owner_payload_from_runtime(runtime),
    )

ursa.util.events.configure_event_logging(*, level=logging.INFO, formatter=None, rich=True)

Configure Python logging for URSA progress events.

The root logger is reset to WARNING and receives one stream handler. The ursa logger is enabled at level so URSA progress remains visible without enabling noisy INFO records from dependencies. When no formatter is supplied, EventConsoleFormatter renders structured event summaries and, optionally, MIME-typed artifacts.

Parameters

level: Logging level for the ursa logger and installed stream handler. formatter: Formatter for the installed handler. When omitted, an EventConsoleFormatter configured for the handler's terminal capability is used. rich: Render artifact bodies with Rich when using the default formatter. This option has no effect when formatter is provided.

Notes

This function clears existing root handlers. Call it once during application startup, or configure logging manually when handler ownership must remain with the embedding application.

Source code in src/ursa/util/events.py

def configure_event_logging(
    *,
    level: int = logging.INFO,
    formatter: logging.Formatter | None = None,
    rich: bool = True,
) -> None:
    """Configure Python logging for URSA progress events.

    The root logger is reset to ``WARNING`` and receives one stream handler.
    The ``ursa`` logger is enabled at ``level`` so URSA progress remains
    visible without enabling noisy ``INFO`` records from dependencies. When no
    formatter is supplied, ``EventConsoleFormatter`` renders structured
    event summaries and, optionally, MIME-typed artifacts.

    Parameters
    ----------
    level:
        Logging level for the ``ursa`` logger and installed stream handler.
    formatter:
        Formatter for the installed handler. When omitted, an
        ``EventConsoleFormatter`` configured for the handler's terminal
        capability is used.
    rich:
        Render artifact bodies with Rich when using the default formatter.
        This option has no effect when ``formatter`` is provided.

    Notes
    -----
    This function clears existing root handlers. Call it once during
    application startup, or configure logging manually when handler ownership
    must remain with the embedding application.
    """
    root_logger = logging.getLogger()
    root_logger.setLevel(logging.WARNING)
    root_logger.handlers.clear()

    # Keep URSA progress visible at the requested level without enabling noisy
    # INFO logs from dependencies such as HTTP clients.
    logging.getLogger("ursa").setLevel(level)

    handler = logging.StreamHandler()
    handler.setLevel(level)
    if formatter is None:
        stream = handler.stream
        is_terminal = bool(hasattr(stream, "isatty") and stream.isatty())
        formatter = EventConsoleFormatter(
            force_terminal=is_terminal,
            render_artifacts=rich,
        )
    handler.setFormatter(formatter)
    root_logger.addHandler(handler)

ursa.util.events.EventLoggingHandler

Bases: BaseCallbackHandler

Write structured URSA progress events to the Python logger.

The handler is intentionally narrow: it only logs URSA's structured custom progress events and ignores all other callback activity. The emitted log line keeps a compact summary for humans while preserving any remaining payload fields as JSON for debugging and ingestion.

Parameters

event_name: Custom event name accepted by the handler. Other custom events are ignored. logger: Logger that receives accepted events. Defaults to the module logger.

Notes

The original structured payload is attached to each log record as ursa_event_payload. Artifact bodies are summarized by MIME type and size rather than copied into the textual log message.

Source code in src/ursa/util/events.py

class EventLoggingHandler(BaseCallbackHandler):
    """Write structured URSA progress events to the Python logger.

    The handler is intentionally narrow: it only logs URSA's structured custom
    progress events and ignores all other callback activity. The emitted log
    line keeps a compact summary for humans while preserving any remaining
    payload fields as JSON for debugging and ingestion.

    Parameters
    ----------
    event_name:
        Custom event name accepted by the handler. Other custom events are
        ignored.
    logger:
        Logger that receives accepted events. Defaults to the module logger.

    Notes
    -----
    The original structured payload is attached to each log record as
    ``ursa_event_payload``. Artifact bodies are summarized by MIME type and
    size rather than copied into the textual log message.
    """

    def __init__(
        self,
        *,
        event_name: str = DEFAULT_EVENT_NAME,
        logger: logging.Logger | None = None,
    ) -> None:
        self.event_name = event_name
        self.logger = logger or LOGGER

    def on_custom_event(
        self,
        name: str,
        data: Any,
        *,
        run_id,
        tags: list[str] | None = None,
        metadata: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> None:
        """Log one matching structured event at ``INFO`` level.

        Events whose name differs from ``event_name`` or whose payload is not
        a dictionary are ignored. ``run_id``, tags, and callback metadata are
        accepted for LangChain callback compatibility but are not included in
        the formatted message.
        """
        if name != self.event_name or not isinstance(data, dict):
            return
        if not self.logger.isEnabledFor(logging.INFO):
            return
        self.logger.info(
            self._format_message(name, data),
            extra={
                "ursa_event_name": name,
                "ursa_event_payload": data,
            },
        )

    def _format_message(self, name: str, data: dict[str, Any]) -> str:
        parts = [f"event={json.dumps(name)}"]
        for key in ("agent", "tool", "name", "stage", "phase", "message"):
            value = data.get(key)
            if value in (None, ""):
                continue
            parts.append(f"{key}={json.dumps(str(value), ensure_ascii=False)}")

        extras = {
            key: value
            for key, value in data.items()
            if key
            not in {
                "agent",
                "artifact",
                "artifacts",
                "tool",
                "name",
                "stage",
                "phase",
                "message",
            }
        }
        artifacts = event_artifacts(data)
        if artifacts:
            extras["artifact_mime_types"] = [
                artifact.get("mime_type") for artifact in artifacts
            ]
        if len(artifacts) == 1:
            artifact = artifacts[0]
            content = artifact.get("content")
            if isinstance(content, str):
                extras["artifact_chars"] = len(content)
            else:
                extras["artifact_type"] = type(content).__name__
        if extras:
            parts.append(
                "data="
                + json.dumps(
                    extras,
                    default=str,
                    ensure_ascii=False,
                    sort_keys=True,
                )
            )
        return " ".join(parts)

on_custom_event(name, data, *, run_id, tags=None, metadata=None, **kwargs)

Log one matching structured event at INFO level.

Events whose name differs from event_name or whose payload is not a dictionary are ignored. run_id, tags, and callback metadata are accepted for LangChain callback compatibility but are not included in the formatted message.

Source code in src/ursa/util/events.py

def on_custom_event(
    self,
    name: str,
    data: Any,
    *,
    run_id,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    **kwargs: Any,
) -> None:
    """Log one matching structured event at ``INFO`` level.

    Events whose name differs from ``event_name`` or whose payload is not
    a dictionary are ignored. ``run_id``, tags, and callback metadata are
    accepted for LangChain callback compatibility but are not included in
    the formatted message.
    """
    if name != self.event_name or not isinstance(data, dict):
        return
    if not self.logger.isEnabledFor(logging.INFO):
        return
    self.logger.info(
        self._format_message(name, data),
        extra={
            "ursa_event_name": name,
            "ursa_event_payload": data,
        },
    )

Artifact helpers and rendering

ursa.util.rendering.event_artifact(content, mime_type, *, metadata=None)

Build a MIME-typed artifact payload.

Parameters

content: Value consumed by the MIME renderer. Keep it serializable when events cross process or persistence boundaries. mime_type: Media type that selects the artifact renderer, for example "text/plain", "text/x-diff", or "application/json". metadata: Optional scalar presentation hints. title and path are recognized by the standard renderer.

Returns

EventArtifact A dictionary suitable for an event's artifact or artifacts field.

Source code in src/ursa/util/rendering.py

def event_artifact(
    content: Any,
    mime_type: str,
    *,
    metadata: Mapping[str, str | float | int] | None = None,
) -> EventArtifact:
    """Build a MIME-typed artifact payload.

    Parameters
    ----------
    content:
        Value consumed by the MIME renderer. Keep it serializable when events
        cross process or persistence boundaries.
    mime_type:
        Media type that selects the artifact renderer, for example
        ``"text/plain"``, ``"text/x-diff"``, or ``"application/json"``.
    metadata:
        Optional scalar presentation hints. ``title`` and ``path`` are
        recognized by the standard renderer.

    Returns
    -------
    EventArtifact
        A dictionary suitable for an event's ``artifact`` or ``artifacts``
        field.
    """
    artifact: EventArtifact = {"content": content, "mime_type": mime_type}
    if metadata:
        artifact["metadata"] = dict(metadata)
    return artifact

ursa.util.rendering.file_artifact(path, *, title='File')

Build an artifact that references a file without reading it.

The payload contains only the path plus its inferred content MIME type. Rendering may later dereference small textual files for syntax-highlighted display; binary, missing, oversized, and undecodable files display as a path instead.

Parameters

path: File path stored in both artifact content and metadata. title: Panel title used by the standard renderer.

Returns

EventArtifact A file-reference artifact with MIME type application/vnd.ursa.file-reference.

Source code in src/ursa/util/rendering.py

def file_artifact(path: str | Path, *, title: str = "File") -> EventArtifact:
    """Build an artifact that references a file without reading it.

    The payload contains only the path plus its inferred content MIME type.
    Rendering may later dereference small textual files for syntax-highlighted
    display; binary, missing, oversized, and undecodable files display as a
    path instead.

    Parameters
    ----------
    path:
        File path stored in both artifact content and metadata.
    title:
        Panel title used by the standard renderer.

    Returns
    -------
    EventArtifact
        A file-reference artifact with MIME type
        ``application/vnd.ursa.file-reference``.
    """
    path = str(path)
    return event_artifact(
        path,
        FILE_REFERENCE_MIME_TYPE,
        metadata={
            "title": title,
            "path": path,
            "content_mime_type": file_content_mime_type(path),
        },
    )

ursa.util.rendering.register_artifact_renderer(mime_type, renderer)

Register or replace the renderer for an artifact MIME type.

Parameters

mime_type: Exact media type used to select the renderer. Parameters such as a charset are removed before lookup. renderer: Callable receiving artifact content and metadata and returning a Rich renderable.

Notes

Registration mutates the process-wide renderer registry. Register custom renderers during application startup.

Source code in src/ursa/util/rendering.py

def register_artifact_renderer(
    mime_type: str, renderer: ArtifactRenderer
) -> None:
    """Register or replace the renderer for an artifact MIME type.

    Parameters
    ----------
    mime_type:
        Exact media type used to select the renderer. Parameters such as a
        charset are removed before lookup.
    renderer:
        Callable receiving artifact content and metadata and returning a Rich
        renderable.

    Notes
    -----
    Registration mutates the process-wide renderer registry. Register custom
    renderers during application startup.
    """
    ARTIFACT_RENDERERS[mime_type] = renderer

ursa.util.rendering.EventConsoleFormatter

Bases: Formatter

Format structured URSA log records for a terminal.

Records carrying an ursa_event_payload dictionary are formatted as a compact source/stage/phase summary. MIME artifacts are appended using the standard Rich renderers. Other records fall back to logging.Formatter behavior.

Parameters

force_terminal: Override Rich terminal detection. Use False for deterministic plain-text output in tests or redirected streams. render_artifacts: Include rendered artifact bodies after the event summary. When False, only the compact summary is returned.

Source code in src/ursa/util/rendering.py

class EventConsoleFormatter(logging.Formatter):
    """Format structured URSA log records for a terminal.

    Records carrying an ``ursa_event_payload`` dictionary are formatted as a
    compact source/stage/phase summary. MIME artifacts are appended using the
    standard Rich renderers. Other records fall back to
    ``logging.Formatter`` behavior.

    Parameters
    ----------
    force_terminal:
        Override Rich terminal detection. Use ``False`` for deterministic
        plain-text output in tests or redirected streams.
    render_artifacts:
        Include rendered artifact bodies after the event summary. When
        ``False``, only the compact summary is returned.
    """

    DETAIL_KEYS = (
        "path",
        "filename",
        "query",
        "output_path",
        "returncode",
        "stdout_chars",
        "stderr_chars",
        "result_chars",
        "error",
    )

    def __init__(
        self,
        *,
        force_terminal: bool | None = None,
        render_artifacts: bool = True,
    ) -> None:
        super().__init__()
        self.force_terminal = force_terminal
        self.render_artifacts = render_artifacts

    def format(self, record: logging.LogRecord) -> str:
        payload = getattr(record, "ursa_event_payload", None)
        if not isinstance(payload, dict):
            return super().format(record)

        source = (
            payload.get("agent")
            or payload.get("tool")
            or payload.get("name")
            or "ursa"
        )
        stage = payload.get("stage")
        phase = payload.get("phase")
        message = payload.get("message") or record.getMessage()
        label = str(source)
        if stage:
            label += f" {stage}"
        if phase:
            label += f"/{phase}"
        details = [
            f"{key}={payload[key]}"
            for key in self.DETAIL_KEYS
            if payload.get(key) not in (None, "")
        ]
        suffix = f" ({', '.join(details)})" if details else ""
        summary = f"[ursa] {label}: {message}{suffix}"
        artifacts = event_artifacts(payload)
        if not self.render_artifacts or not artifacts:
            return summary

        output = StringIO()
        console = Console(
            file=output,
            color_system="auto",
            force_terminal=self.force_terminal,
        )
        console.print(render_event_artifacts(artifacts))
        return f"{summary}\n{output.getvalue().rstrip()}"

ursa.util.rendering.render_event_artifact(artifact)

Render one artifact as a titled Rich panel.

A registered MIME renderer produces the panel body. Unknown media types fall back to syntax- or plain-text rendering. File-reference artifacts are dereferenced only when they identify a readable textual file no larger than MAX_INLINE_FILE_BYTES.

Parameters

artifact: Mapping containing content, mime_type, and optional metadata.

Returns

RenderableType A Rich panel ready for a console, table, or callback interface.

Source code in src/ursa/util/rendering.py

def render_event_artifact(artifact: Mapping[str, Any]) -> RenderableType:
    """Render one artifact as a titled Rich panel.

    A registered MIME renderer produces the panel body. Unknown media types
    fall back to syntax- or plain-text rendering. File-reference artifacts are
    dereferenced only when they identify a readable textual file no larger
    than ``MAX_INLINE_FILE_BYTES``.

    Parameters
    ----------
    artifact:
        Mapping containing ``content``, ``mime_type``, and optional metadata.

    Returns
    -------
    RenderableType
        A Rich panel ready for a console, table, or callback interface.
    """
    raw_content = artifact.get("content", "")
    mime_type = str(artifact.get("mime_type") or "text/plain").split(";", 1)[0]
    metadata = artifact.get("metadata")
    if not isinstance(metadata, Mapping):
        metadata = {}
    path = str(metadata.get("path") or "")
    title = str(metadata.get("title") or path or mime_type)

    if renderer := ARTIFACT_RENDERERS.get(mime_type):
        body = renderer(raw_content, metadata)
    else:
        content = str(raw_content)
        lexer = {"text/x-diff": "diff", "text/x-python": "python"}.get(
            mime_type
        )
        if lexer is None:
            try:
                lexer = str(Syntax.guess_lexer(path or "artifact.txt", content))
            except (ClassNotFound, TypeError, ValueError):
                lexer = "text"
        body = Syntax(content, lexer, line_numbers=False, word_wrap=True)
    return Panel(body, title=title, border_style="green", expand=False)

ursa.util.rendering.render_event_artifacts(artifacts)

Render artifacts as one panel or an equal-width horizontal row.

Parameters

artifacts: Ordered artifact mappings to render.

Returns

RenderableType The single artifact panel, or a Rich table that constrains multiple panels to side-by-side columns.

Notes

Passing an empty list produces an empty Rich table. Event consumers normally call this function only after confirming artifacts are present.

Source code in src/ursa/util/rendering.py

def render_event_artifacts(
    artifacts: list[Mapping[str, Any]],
) -> RenderableType:
    """Render artifacts as one panel or an equal-width horizontal row.

    Parameters
    ----------
    artifacts:
        Ordered artifact mappings to render.

    Returns
    -------
    RenderableType
        The single artifact panel, or a Rich table that constrains multiple
        panels to side-by-side columns.

    Notes
    -----
    Passing an empty list produces an empty Rich table. Event consumers
    normally call this function only after confirming artifacts are present.
    """
    rendered = [render_event_artifact(artifact) for artifact in artifacts]
    if len(rendered) == 1:
        return rendered[0]

    artifact_table = Table.grid(expand=True, padding=(0, 1))
    for _ in rendered:
        artifact_table.add_column(ratio=1, overflow="fold")
    artifact_table.add_row(*rendered)
    return artifact_table