Skip to content

cayu.observability.hooks

Public Cayu API. Definitions are grouped by their public source module, and definitions from private implementation modules appear on their package’s page; adding a symbol does not move existing links. The import shown at each entry is a supported public namespace. Read the Python API overview and install the required extra.

from cayu.observability import AfterToolCallDecision

An after_tool_call hook’s instruction for the result it just inspected.

pass_through leaves the result unchanged; modify replaces it with modified_result before the result enters the transcript. Returning None from the hook equals pass_through.

Bases: BaseModel.

Defaults below are source expressions; Field(...) includes schema/validation settings and default_factory creates a fresh value. Inherited declarations are included. Constructor behavior follows the model or dataclass shown in the bases/decorators.

Name Type Default
action Literal['pass_through', 'modify'] required
modified_result ToolResult | None None

Release source · Runtime guarantees · Capability guides

from cayu.observability import BeforeToolCallDecision

A before_tool_call hook’s instruction for the call it just inspected.

proceed runs the tool unchanged; proceed_modified runs it with modified_arguments; short_circuit skips the tool and uses synthetic_result; block skips the tool and returns an error result carrying block_reason. Returning None from the hook equals proceed. Modified arguments are validated and copied as portable durable JSON at construction.

Bases: BaseModel.

Defaults below are source expressions; Field(...) includes schema/validation settings and default_factory creates a fresh value. Inherited declarations are included. Constructor behavior follows the model or dataclass shown in the bases/decorators.

Name Type Default
action Literal['proceed', 'proceed_modified', 'short_circuit', 'block'] required
modified_arguments dict[str, Any] | None None
synthetic_result ToolResult | None None
block_reason str | None None

Release source · Runtime guarantees · Capability guides

from cayu.observability import BeforeToolCallHookContext

The before_tool_call view of a call: identity + arguments, no result yet.

Distinct from ToolCallHookContext because before execution there is no result or result event. arguments is a read-only copy; mutating it is a no-op — a before-hook changes the call only by returning a BeforeToolCallDecision. Arguments satisfy the portable durable JSON contract before the hook can observe them.

Bases: _HookActionContext.

BeforeToolCallHookContext.__init__(*, runtime: RuntimeHookRuntime, hook_name: str, phase: RuntimeHookPhase, session: Session, tool_name: str, tool_call_id: str, arguments: dict[str, Any], task_id: str | None, publication_actions_allowed: bool=True, execution_profile: ExecutionProfileIdentity | None=None) -> None
BeforeToolCallHookContext.hook_name() -> str
BeforeToolCallHookContext.phase() -> RuntimeHookPhase
BeforeToolCallHookContext.session() -> Session
BeforeToolCallHookContext.execution_profile() -> ExecutionProfileIdentity | None

Return the exact immutable profile resolved for this invocation.

BeforeToolCallHookContext.actions() -> list[dict[str, Any]]
async BeforeToolCallHookContext.fork_session(request: ForkSessionRequest) -> list[Event]
async BeforeToolCallHookContext.dispatch(request: DispatchRequest) -> DispatchHandle
async BeforeToolCallHookContext.dispatch_inline(request: DispatchRequest) -> list[Event]
async BeforeToolCallHookContext.create_task(request: TaskCreate) -> Task
async BeforeToolCallHookContext.emit_custom_event(event_type: str, *, session_id: str | None=None, payload: dict[str, Any] | None=None) -> Event
BeforeToolCallHookContext.tool_name() -> str
BeforeToolCallHookContext.tool_call_id() -> str
BeforeToolCallHookContext.arguments() -> dict[str, Any]
BeforeToolCallHookContext.task_id() -> str | None

Release source · Runtime guarantees · Capability guides

from cayu.observability import RuntimeHook
RuntimeHook.execution_profile_identity() -> ExecutionProfileBehaviorIdentity | None

Return a stable application declaration, or None when non-portable.

RuntimeHook.name() -> str
async RuntimeHook.after_session_completed(context: RuntimeHookContext) -> None

Run after a session reaches completed state.

async RuntimeHook.after_session_failed(context: RuntimeHookContext) -> None

Run after a session reaches failed state.

async RuntimeHook.after_session_interrupted(context: RuntimeHookContext) -> None

Run after a session reaches interrupted state.

async RuntimeHook.before_tool_call(context: BeforeToolCallHookContext) -> BeforeToolCallDecision | None

Run after policy authorization, before the tool executes.

Return None to proceed unchanged, or a BeforeToolCallDecision to modify the arguments, short-circuit with a synthetic result, or block the call.

async RuntimeHook.after_tool_call(context: ToolCallHookContext) -> AfterToolCallDecision | None

Run after the tool executes, before its result enters the transcript.

Return None to pass the result through unchanged, or an AfterToolCallDecision to replace it.

Release source · Runtime guarantees · Capability guides

from cayu.observability import RuntimeHookContext

Bases: _HookActionContext.

RuntimeHookContext.__init__(*, runtime: RuntimeHookRuntime, hook_name: str, phase: RuntimeHookPhase, session: Session, terminal_event: Event, execution_profile: ExecutionProfileIdentity | None=None) -> None
RuntimeHookContext.hook_name() -> str
RuntimeHookContext.phase() -> RuntimeHookPhase
RuntimeHookContext.session() -> Session
RuntimeHookContext.execution_profile() -> ExecutionProfileIdentity | None

Return the exact immutable profile resolved for this invocation.

RuntimeHookContext.actions() -> list[dict[str, Any]]
async RuntimeHookContext.fork_session(request: ForkSessionRequest) -> list[Event]
async RuntimeHookContext.dispatch(request: DispatchRequest) -> DispatchHandle
async RuntimeHookContext.dispatch_inline(request: DispatchRequest) -> list[Event]
async RuntimeHookContext.create_task(request: TaskCreate) -> Task
async RuntimeHookContext.emit_custom_event(event_type: str, *, session_id: str | None=None, payload: dict[str, Any] | None=None) -> Event
RuntimeHookContext.terminal_event() -> Event

Release source · Runtime guarantees · Capability guides

from cayu.observability import RuntimeHookPhase

Bases: StrEnum.

Defaults below are source expressions; Field(...) includes schema/validation settings and default_factory creates a fresh value. Inherited declarations are included. Constructor behavior follows the model or dataclass shown in the bases/decorators.

Name Type Default
AFTER_SESSION_COMPLETED class attribute 'after_session_completed'
AFTER_SESSION_FAILED class attribute 'after_session_failed'
AFTER_SESSION_INTERRUPTED class attribute 'after_session_interrupted'
BEFORE_TOOL_CALL class attribute 'before_tool_call'
AFTER_TOOL_CALL class attribute 'after_tool_call'

Release source · Runtime guarantees · Capability guides

from cayu.observability import ToolCallHookContext

Bases: _HookActionContext.

ToolCallHookContext.__init__(*, runtime: RuntimeHookRuntime, hook_name: str, phase: RuntimeHookPhase, session: Session, tool_event: Event, tool_name: str, tool_call_id: str, arguments: dict[str, Any], result: ToolResult, task_id: str | None, publication_actions_allowed: bool=True, execution_profile: ExecutionProfileIdentity | None=None) -> None
ToolCallHookContext.hook_name() -> str
ToolCallHookContext.phase() -> RuntimeHookPhase
ToolCallHookContext.session() -> Session
ToolCallHookContext.execution_profile() -> ExecutionProfileIdentity | None

Return the exact immutable profile resolved for this invocation.

ToolCallHookContext.actions() -> list[dict[str, Any]]
async ToolCallHookContext.fork_session(request: ForkSessionRequest) -> list[Event]
async ToolCallHookContext.dispatch(request: DispatchRequest) -> DispatchHandle
async ToolCallHookContext.dispatch_inline(request: DispatchRequest) -> list[Event]
async ToolCallHookContext.create_task(request: TaskCreate) -> Task
async ToolCallHookContext.emit_custom_event(event_type: str, *, session_id: str | None=None, payload: dict[str, Any] | None=None) -> Event
ToolCallHookContext.tool_event() -> Event
ToolCallHookContext.tool_name() -> str
ToolCallHookContext.tool_call_id() -> str
ToolCallHookContext.arguments() -> dict[str, Any]
ToolCallHookContext.result() -> ToolResult
ToolCallHookContext.task_id() -> str | None

Release source · Runtime guarantees · Capability guides