Python API: Hooks

Your code around the agent loop: before and after model and tool calls. Generated from the docstrings and type hints of criogenai 0.1.7. See also: Python API overview, the same module in TypeScript.

Hooks: your code at fixed points of the agent loop.

Pass hooks with Agent(hooks=[...]). A hook is any object with some of these methods (each may be sync or async):

• on_start(ctx): once per run segment; may return HookAction(state=…).
• before_model(ctx): before each model call; may replace the transcript (HookAction(messages=[…])) or end the run.
• after_model(ctx): after each model reply (ctx.messages ends with it); may replace it (HookAction(message={…})) or end the run.
• before_tool(call): before each tool call; may return a changed ToolCall, a ToolResult to skip the tool, or raise ToolError.
• after_tool(call, result): after each tool call; may return a replacement result.
• on_end(ctx, outcome): always, once per segment ("completed", "suspended" or "failed"); cannot change the outcome.

on_start, before_model and before_tool run in list order, the others in reverse. An exception in on_start, before_model or after_model fails the run with hook_failed (the exception is its __cause__); one in before_tool / after_tool counts as the tool's (see on_tool_exception). Hook methods run on your event loop: keep sync ones quick. A hook's name attribute, or else its class name, identifies its saved state.

Example:

from criogenai import Agent, FakeModel, HookAction

class StopAfterTwoCalls:
    def before_model(self, ctx):
        if ctx.model_calls >= 2:
            return HookAction(end=True, output="stopped: too many model calls")
        return None

agent = Agent(FakeModel(["Hello."]), hooks=[StopAfterTwoCalls()])

from criogenai import HookContext, HookAction, ToolCall, ToolResult

Contents#

HookContext#

@dataclass
class HookContext:
    run_id: str
    messages: List[Dict[str, Any]]
    model_calls: int
    hook_state: Any
    state: Dict[str, Any]

What a hook sees: passed to on_start, before_model, after_model and on_end.

Attributes

• run_id (str): The run's id.
• messages (List[Dict[str, Any]]): The transcript so far, as dicts (the form of result.messages). In after_model it ends with the new reply.
• model_calls (int): Model calls made in the run so far.
• hook_state (Any): What this hook saved with HookAction(state=...), or None.
• state (Dict[str, Any]): The whole run state, including values passed as keyword arguments to run or send. Read-only: changes are not kept.

HookAction#

@dataclass
class HookAction:
    state: Any = _UNSET
    messages: Optional[List[Dict[str, Any]]] = None
    message: Optional[Dict[str, Any]] = None
    end: bool = False
    output: Any = None

What a hook wants done; returned by on_start, before_model or after_model (return None to change nothing).

Attributes

• state (Any, default _UNSET): A JSON value to save for this hook; later calls see it as ctx.hook_state. Left unset, the saved state is kept.
• messages (Optional[List[Dict[str, Any]]], default None): A new transcript to send to the model instead (before_model).
• message (Optional[Dict[str, Any]], default None): A new reply to use instead of the model's (after_model), as a message dict.
• end (bool, default False): End the run now, with stop reason "hook_end".
• output (Any, default None): With end, the run's output.

Example

class CountRuns:
    def on_start(self, ctx):
        runs = (ctx.hook_state or {}).get("runs", 0) + 1
        return HookAction(state={"runs": runs})

ToolCall#

@dataclass
class ToolCall:
    call_id: str
    name: str
    args: Dict[str, Any] = field(default_factory=dict)

A tool call, as before_tool and after_tool see it.

In before_tool, change name or args and return the call to run the changed call.

Attributes

• call_id (str): The call's id.
• name (str): The tool to call.
• args (Dict[str, Any], default field(default_factory=dict)): The arguments, as a dict.

ToolResult#

@dataclass
class ToolResult:
    value: Any

Returned by before_tool to skip the tool and use value as its result, for example a cached answer.

Attributes

• value (Any): The result the model receives, any JSON value.

Example

class Cache:
    def before_tool(self, call):
        if call.name == "weather" and call.args.get("city") == "Paris":
            return ToolResult("sunny (cached)")
        return None