Hooks

Contract

HookProvider is the base class every hook subclasses; HookRegistry collects providers and dispatches events in priority order. HookPriority is a plain class of integer constants that determine dispatch order (lower = earlier — e.g. a guardrail or policy hook runs before a logging hook). ProtectedEvent marks events whose payloads can be mutated by hooks (model calls, tool calls) versus the read-only lifecycle events.

HookProvider

Bases: ABC

Abstract base class for hook providers.

Hook providers implement lifecycle callbacks that are invoked during agent execution. Multiple providers can be registered, with execution order determined by priority (lower = earlier).

Example

class MyLoggingHook(HookProvider): @property def priority(self) -> int: return HookPriority.OBSERVABILITY_DEFAULT

async def on_before_invocation(
    self, prompt: str, state: AgentState
) -> AgentState:
    print(f"Starting: {prompt[:50]}...")
    return state

async def on_after_invocation(
    self, state: AgentState, success: bool
) -> None:
    print(f"Completed: success={success}")

priority abstractmethod property

priority: int

Hook priority (lower = earlier execution).

Use HookPriority constants for standard ranges.

name property

name: str

Hook provider name for identification.

on_before_invocation async

on_before_invocation(prompt: str, state: AgentState) -> AgentState

Called before agent starts processing.

Parameters:

Name	Type	Description	Default
prompt	str	The user prompt being processed	required
state	AgentState	Current agent state	required

Returns:

Type	Description
AgentState	Potentially modified agent state

Source code in .sdk/src/tulip/hooks/provider.py

async def on_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Called before agent starts processing.

    Args:
        prompt: The user prompt being processed
        state: Current agent state

    Returns:
        Potentially modified agent state
    """
    return state

on_after_invocation async

on_after_invocation(state: AgentState, success: bool) -> None

Called after agent completes processing.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution completed successfully	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """Called after agent completes processing.

    Args:
        state: Final agent state
        success: Whether execution completed successfully
    """

on_before_tool_call async

on_before_tool_call(event: BeforeToolCallEvent) -> None

Called before tool execution.

Modify event.arguments to change tool inputs. Set event.cancel = True or a string reason to skip execution. event.tool_name and event.tool_call_id are read-only.

Parameters:

Name	Type	Description	Default
event	BeforeToolCallEvent	Write-protected event. Writable: arguments, cancel.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_before_tool_call(
    self,
    event: BeforeToolCallEvent,
) -> None:
    """Called before tool execution.

    Modify event.arguments to change tool inputs.
    Set event.cancel = True or a string reason to skip execution.
    event.tool_name and event.tool_call_id are read-only.

    Args:
        event: Write-protected event. Writable: arguments, cancel.
    """

on_after_tool_call async

on_after_tool_call(event: AfterToolCallEvent) -> None

Called after tool execution.

Set event.retry = True to re-execute the tool. Set event.result to replace the tool result. event.tool_name and event.error are read-only.

Parameters:

Name	Type	Description	Default
event	AfterToolCallEvent	Write-protected event. Writable: result, retry.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_tool_call(
    self,
    event: AfterToolCallEvent,
) -> None:
    """Called after tool execution.

    Set event.retry = True to re-execute the tool.
    Set event.result to replace the tool result.
    event.tool_name and event.error are read-only.

    Args:
        event: Write-protected event. Writable: result, retry.
    """

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Called at the start of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the start of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

Called at the end of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the end of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_before_model_call async

on_before_model_call(event: BeforeModelCallEvent) -> None

Called before each model.complete() call.

Modify event.messages to change what the model sees. event.tools is read-only (inspect only).

Parameters:

Name	Type	Description	Default
event	BeforeModelCallEvent	Write-protected event. Writable: messages.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_before_model_call(
    self,
    event: BeforeModelCallEvent,
) -> None:
    """Called before each model.complete() call.

    Modify event.messages to change what the model sees.
    event.tools is read-only (inspect only).

    Args:
        event: Write-protected event. Writable: messages.
    """

on_after_model_call async

on_after_model_call(event: AfterModelCallEvent) -> None

Called after each model.complete() call.

Set event.retry = True to discard response and re-call. Set event.response to replace the response. event.messages is read-only.

Parameters:

Name	Type	Description	Default
event	AfterModelCallEvent	Write-protected event. Writable: response, retry.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_model_call(
    self,
    event: AfterModelCallEvent,
) -> None:
    """Called after each model.complete() call.

    Set event.retry = True to discard response and re-call.
    Set event.response to replace the response.
    event.messages is read-only.

    Args:
        event: Write-protected event. Writable: response, retry.
    """

register_hooks

register_hooks() -> dict[str, bool]

Return which hooks this provider implements.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping hook names to whether they are implemented.
dict[str, bool]	Useful for optimization - registry can skip calling unimplemented hooks.

Source code in .sdk/src/tulip/hooks/provider.py

def register_hooks(self) -> dict[str, bool]:
    """Return which hooks this provider implements.

    Returns:
        Dictionary mapping hook names to whether they are implemented.
        Useful for optimization - registry can skip calling unimplemented hooks.
    """
    return {
        "on_before_invocation": True,
        "on_after_invocation": True,
        "on_before_tool_call": True,
        "on_after_tool_call": True,
        "on_iteration_start": True,
        "on_iteration_end": True,
        "on_before_model_call": True,
        "on_after_model_call": True,
    }

HookPriority

Standard priority ranges for hook ordering.

Lower priority = earlier execution.

Ranges: - SECURITY (0-99): Security checks, input validation, rate limiting - OBSERVABILITY (100-199): Logging, metrics, tracing - BUSINESS (200-299): Business logic, custom transformations - DEFAULT (300+): General purpose hooks

ProtectedEvent

Base class for write-protected hook events.

Subclasses declare _writable as a set of field names that hooks may modify. All other attributes are read-only after init.

Setting a read-only field raises AttributeError with a clear message.

Example

class BeforeToolCallEvent(ProtectedEvent): _writable = {"arguments", "cancel", "cancel_reason"}

def __init__(self, tool_name, arguments):
    self._init("tool_name", tool_name)
    self._init("arguments", arguments)
    self._init("cancel", False)
    self._init("cancel_reason", "")

__setattr__

__setattr__(name: str, value: Any) -> None

Only allow setting writable fields.

Source code in .sdk/src/tulip/hooks/provider.py

def __setattr__(self, name: str, value: Any) -> None:
    """Only allow setting writable fields."""
    if name.startswith("_") or name in self._writable:
        object.__setattr__(self, name, value)
    else:
        writable = ", ".join(sorted(self._writable)) or "none"
        msg = (
            f"Cannot set '{name}' on {type(self).__name__} — "
            f"read-only. Writable fields: {writable}"
        )
        raise AttributeError(msg)

HookRegistry

HookRegistry()

Registry for managing hook providers.

The registry maintains a priority-ordered list of hook providers and dispatches lifecycle events to them in order.

Example

registry = HookRegistry() registry.add_provider(LoggingHook()) registry.add_provider(GuardrailsHook())

During agent execution

state = await registry.emit_before_invocation(prompt, state)

... agent runs ...

await registry.emit_after_invocation(state, success=True)

Initialize empty hook registry.

Source code in .sdk/src/tulip/hooks/registry.py

def __init__(self) -> None:
    """Initialize empty hook registry."""
    self._providers: list[HookProvider] = []
    self._sorted = True

providers property

providers: list[HookProvider]

Get all registered providers in priority order.

add_provider

add_provider(provider: HookProvider) -> None

Register a hook provider.

Parameters:

Name	Type	Description	Default
provider	HookProvider	Hook provider to register	required

Raises:

Type	Description
ValueError	If provider with same name already registered

Source code in .sdk/src/tulip/hooks/registry.py

def add_provider(self, provider: HookProvider) -> None:
    """Register a hook provider.

    Args:
        provider: Hook provider to register

    Raises:
        ValueError: If provider with same name already registered
    """
    for existing in self._providers:
        if existing.name == provider.name:
            msg = f"Hook provider '{provider.name}' already registered"
            raise ValueError(msg)

    self._providers.append(provider)
    self._sorted = False
    logger.debug(
        "Registered hook provider '%s' with priority %d",
        provider.name,
        provider.priority,
    )

remove_provider

remove_provider(name: str) -> bool

Remove a hook provider by name.

Parameters:

Name	Type	Description	Default
name	str	Name of the provider to remove	required

Returns:

Type	Description
bool	True if provider was removed, False if not found

Source code in .sdk/src/tulip/hooks/registry.py

def remove_provider(self, name: str) -> bool:
    """Remove a hook provider by name.

    Args:
        name: Name of the provider to remove

    Returns:
        True if provider was removed, False if not found
    """
    for i, provider in enumerate(self._providers):
        if provider.name == name:
            self._providers.pop(i)
            logger.debug("Removed hook provider '%s'", name)
            return True
    return False

get_provider

get_provider(name: str) -> HookProvider | None

Get a hook provider by name.

Parameters:

Name	Type	Description	Default
name	str	Name of the provider to find	required

Returns:

Type	Description
HookProvider | None	The provider if found, None otherwise

Source code in .sdk/src/tulip/hooks/registry.py

def get_provider(self, name: str) -> HookProvider | None:
    """Get a hook provider by name.

    Args:
        name: Name of the provider to find

    Returns:
        The provider if found, None otherwise
    """
    for provider in self._providers:
        if provider.name == name:
            return provider
    return None

__len__

__len__() -> int

Return number of registered providers.

Source code in .sdk/src/tulip/hooks/registry.py

def __len__(self) -> int:
    """Return number of registered providers."""
    return len(self._providers)

__contains__

__contains__(name: str) -> bool

Check if a provider with given name is registered.

Source code in .sdk/src/tulip/hooks/registry.py

def __contains__(self, name: str) -> bool:
    """Check if a provider with given name is registered."""
    return any(p.name == name for p in self._providers)

emit_before_invocation async

emit_before_invocation(prompt: str, state: AgentState) -> AgentState

Emit before_invocation event to all providers.

Parameters:

Name	Type	Description	Default
prompt	str	User prompt being processed	required
state	AgentState	Current agent state	required

Returns:

Type	Description
AgentState	Potentially modified agent state

Source code in .sdk/src/tulip/hooks/registry.py

async def emit_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Emit before_invocation event to all providers.

    Args:
        prompt: User prompt being processed
        state: Current agent state

    Returns:
        Potentially modified agent state
    """
    self._ensure_sorted()
    for provider in self._providers:
        try:
            state = await provider.on_before_invocation(prompt, state)
        except Exception:
            logger.exception(
                "Error in hook provider '%s' on_before_invocation",
                provider.name,
            )
            raise
    return state

emit_after_invocation async

emit_after_invocation(state: AgentState, success: bool) -> None

Emit after_invocation event to all providers.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution completed successfully	required

Source code in .sdk/src/tulip/hooks/registry.py

async def emit_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """Emit after_invocation event to all providers.

    Args:
        state: Final agent state
        success: Whether execution completed successfully
    """
    self._ensure_sorted()
    errors: list[tuple[str, Exception]] = []
    # Reverse order: last-registered-first for proper teardown
    for provider in reversed(self._providers):
        try:
            await provider.on_after_invocation(state, success)
        except Exception as e:
            logger.exception(
                "Error in hook provider '%s' on_after_invocation",
                provider.name,
            )
            errors.append((provider.name, e))

    # Re-raise first error if any occurred
    if errors:
        name, error = errors[0]
        msg = f"Hook provider '{name}' failed in on_after_invocation: {error}"
        raise RuntimeError(msg) from error

emit_before_tool_call async

emit_before_tool_call(tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]

Emit before_tool_call event to all providers.

Parameters:

Name	Type	Description	Default
tool_name	str	Name of the tool being called	required
arguments	dict[str, Any]	Tool arguments	required

Returns:

Type	Description
dict[str, Any]	Potentially modified arguments

Source code in .sdk/src/tulip/hooks/registry.py

async def emit_before_tool_call(
    self,
    tool_name: str,
    arguments: dict[str, Any],
) -> dict[str, Any]:
    """Emit before_tool_call event to all providers.

    Args:
        tool_name: Name of the tool being called
        arguments: Tool arguments

    Returns:
        Potentially modified arguments
    """
    from tulip.hooks.provider import BeforeToolCallEvent

    self._ensure_sorted()
    event = BeforeToolCallEvent(tool_name=tool_name, tool_call_id="", arguments=arguments)
    for provider in self._providers:
        try:
            await provider.on_before_tool_call(event)
        except Exception:
            logger.exception(
                "Error in hook provider '%s' on_before_tool_call",
                provider.name,
            )
            raise
    modified_arguments: dict[str, Any] = event.arguments
    return modified_arguments

emit_after_tool_call async

emit_after_tool_call(tool_name: str, result: Any, error: str | None, *, tool_call_id: str = '', arguments: dict[str, Any] | None = None) -> None

Emit after_tool_call event to all providers.

Parameters:

Name	Type	Description	Default
tool_name	str	Name of the tool that was called	required
result	Any	Tool result (if successful)	required
error	str | None	Error message (if failed)	required
tool_call_id	str	ID of the tool call (correlates with BeforeToolCallEvent).	''
arguments	dict[str, Any] | None	Arguments the tool was invoked with (post-hook mutation).	None

Source code in .sdk/src/tulip/hooks/registry.py

async def emit_after_tool_call(
    self,
    tool_name: str,
    result: Any,
    error: str | None,
    *,
    tool_call_id: str = "",
    arguments: dict[str, Any] | None = None,
) -> None:
    """Emit after_tool_call event to all providers.

    Args:
        tool_name: Name of the tool that was called
        result: Tool result (if successful)
        error: Error message (if failed)
        tool_call_id: ID of the tool call (correlates with BeforeToolCallEvent).
        arguments: Arguments the tool was invoked with (post-hook mutation).
    """
    from tulip.hooks.provider import AfterToolCallEvent

    self._ensure_sorted()
    event = AfterToolCallEvent(
        tool_name=tool_name,
        result=result,
        error=error,
        tool_call_id=tool_call_id,
        arguments=arguments,
    )
    errors: list[tuple[str, Exception]] = []
    # Reverse order for proper teardown
    for provider in reversed(self._providers):
        try:
            await provider.on_after_tool_call(event)
        except Exception as e:
            logger.exception(
                "Error in hook provider '%s' on_after_tool_call",
                provider.name,
            )
            errors.append((provider.name, e))

    if errors:
        name, error_exc = errors[0]
        msg = f"Hook provider '{name}' failed in on_after_tool_call: {error_exc}"
        raise RuntimeError(msg) from error_exc

emit_iteration_start async

emit_iteration_start(iteration: int, state: AgentState) -> None

Emit iteration_start event to all providers.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number	required
state	AgentState	Current agent state	required

Source code in .sdk/src/tulip/hooks/registry.py

async def emit_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Emit iteration_start event to all providers.

    Args:
        iteration: Current iteration number
        state: Current agent state
    """
    self._ensure_sorted()
    tasks = [provider.on_iteration_start(iteration, state) for provider in self._providers]
    if tasks:
        await asyncio.gather(*tasks, return_exceptions=True)

emit_iteration_end async

emit_iteration_end(iteration: int, state: AgentState) -> None

Emit iteration_end event to all providers.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number	required
state	AgentState	Current agent state	required

Source code in .sdk/src/tulip/hooks/registry.py

async def emit_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Emit iteration_end event to all providers.

    Args:
        iteration: Current iteration number
        state: Current agent state
    """
    self._ensure_sorted()
    tasks = [provider.on_iteration_end(iteration, state) for provider in self._providers]
    if tasks:
        await asyncio.gather(*tasks, return_exceptions=True)

emit_before_model_call async

emit_before_model_call(messages: list[Any], tools: list[dict[str, Any]] | None) -> list[Any]

Emit before_model_call event to all providers.

Parameters:

Name	Type	Description	Default
messages	list[Any]	Messages about to be sent to the model	required
tools	list[dict[str, Any]] | None	Tool schemas (if any)	required

Returns:

Type	Description
list[Any]	Potentially modified messages list

Source code in .sdk/src/tulip/hooks/registry.py

async def emit_before_model_call(
    self,
    messages: list[Any],
    tools: list[dict[str, Any]] | None,
) -> list[Any]:
    """Emit before_model_call event to all providers.

    Args:
        messages: Messages about to be sent to the model
        tools: Tool schemas (if any)

    Returns:
        Potentially modified messages list
    """
    from tulip.hooks.provider import BeforeModelCallEvent

    self._ensure_sorted()
    event = BeforeModelCallEvent(messages=messages, tools=tools)
    for provider in self._providers:
        try:
            await provider.on_before_model_call(event)
        except Exception:
            logger.exception(
                "Error in hook provider '%s' on_before_model_call",
                provider.name,
            )
            raise
    messages_out: list[Any] = event.messages
    return messages_out

emit_after_model_call async

emit_after_model_call(response: Any, messages: list[Any]) -> Any

Emit after_model_call event to all providers.

Parameters:

Name	Type	Description	Default
response	Any	The ModelResponse from the model	required
messages	list[Any]	The messages that were sent	required

Returns:

Type	Description
Any	Potentially modified response

Source code in .sdk/src/tulip/hooks/registry.py

async def emit_after_model_call(
    self,
    response: Any,
    messages: list[Any],
) -> Any:
    """Emit after_model_call event to all providers.

    Args:
        response: The ModelResponse from the model
        messages: The messages that were sent

    Returns:
        Potentially modified response
    """
    from tulip.hooks.provider import AfterModelCallEvent

    self._ensure_sorted()
    event = AfterModelCallEvent(response=response, messages=messages)
    # Reverse order for proper teardown
    for provider in reversed(self._providers):
        try:
            await provider.on_after_model_call(event)
        except Exception:
            logger.exception(
                "Error in hook provider '%s' on_after_model_call",
                provider.name,
            )
            raise
    return event.response

emit async

emit(event_name: str, *args: Any, **kwargs: Any) -> Any

Generic event emission for custom hook points.

Parameters:

Name	Type	Description	Default
event_name	str	Name of the hook method to call	required
*args	Any	Positional arguments to pass	()
**kwargs	Any	Keyword arguments to pass	{}

Returns:

Type	Description
Any	Result from the last provider that returned a non-None value

Source code in .sdk/src/tulip/hooks/registry.py

async def emit(
    self,
    event_name: str,
    *args: Any,
    **kwargs: Any,
) -> Any:
    """Generic event emission for custom hook points.

    Args:
        event_name: Name of the hook method to call
        *args: Positional arguments to pass
        **kwargs: Keyword arguments to pass

    Returns:
        Result from the last provider that returned a non-None value
    """
    self._ensure_sorted()
    result = None
    for provider in self._providers:
        method = getattr(provider, event_name, None)
        if method is not None and callable(method):
            try:
                ret = await method(*args, **kwargs)
                if ret is not None:
                    result = ret
            except Exception:
                logger.exception(
                    "Error in hook provider '%s' %s",
                    provider.name,
                    event_name,
                )
                raise
    return result

create_registry

create_registry(*providers: HookProvider) -> HookRegistry

Create a registry with the given providers.

Parameters:

Name	Type	Description	Default
*providers	HookProvider	Hook providers to register	()

Returns:

Type	Description
HookRegistry	New HookRegistry with all providers registered

Example

registry = create_registry( LoggingHook(), TelemetryHook(), GuardrailsHook(), )

Source code in .sdk/src/tulip/hooks/registry.py

def create_registry(*providers: HookProvider) -> HookRegistry:
    """Create a registry with the given providers.

    Args:
        *providers: Hook providers to register

    Returns:
        New HookRegistry with all providers registered

    Example:
        registry = create_registry(
            LoggingHook(),
            TelemetryHook(),
            GuardrailsHook(),
        )
    """
    registry = HookRegistry()
    for provider in providers:
        registry.add_provider(provider)
    return registry

Hook events

Write-protected events

Hooks observing these events can mutate the payload before it reaches the model / tool / next stage.

BeforeModelCallEvent

BeforeModelCallEvent(messages: list[Any], tools: list[Any] | None)

Bases: ProtectedEvent

Event fired before each model.complete() call.

Writable fields

messages: Modify/trim messages before they reach the model.

Read-only fields

tools: Tool schemas (inspect only).

Example

async def on_before_model_call(self, event): # Trim to last 10 messages to fit context window event.messages = event.messages[-10:]

Source code in .sdk/src/tulip/hooks/provider.py

def __init__(self, messages: list[Any], tools: list[Any] | None) -> None:
    self._init("messages", messages)
    self._init("tools", tools)

__setattr__

__setattr__(name: str, value: Any) -> None

Only allow setting writable fields.

Source code in .sdk/src/tulip/hooks/provider.py

def __setattr__(self, name: str, value: Any) -> None:
    """Only allow setting writable fields."""
    if name.startswith("_") or name in self._writable:
        object.__setattr__(self, name, value)
    else:
        writable = ", ".join(sorted(self._writable)) or "none"
        msg = (
            f"Cannot set '{name}' on {type(self).__name__} — "
            f"read-only. Writable fields: {writable}"
        )
        raise AttributeError(msg)

AfterModelCallEvent

AfterModelCallEvent(response: Any, messages: list[Any])

Bases: ProtectedEvent

Event fired after each model.complete() call.

Writable fields

retry: Set True to discard response and re-call the model. response: Replace the model response.

Read-only fields

messages: The messages that were sent.

Example

async def on_after_model_call(self, event): if not event.response.message.content: event.retry = True # Empty response, retry

Source code in .sdk/src/tulip/hooks/provider.py

def __init__(self, response: Any, messages: list[Any]) -> None:
    self._init("response", response)
    self._init("messages", messages)
    self._init("retry", False)

__setattr__

__setattr__(name: str, value: Any) -> None

Only allow setting writable fields.

Source code in .sdk/src/tulip/hooks/provider.py

def __setattr__(self, name: str, value: Any) -> None:
    """Only allow setting writable fields."""
    if name.startswith("_") or name in self._writable:
        object.__setattr__(self, name, value)
    else:
        writable = ", ".join(sorted(self._writable)) or "none"
        msg = (
            f"Cannot set '{name}' on {type(self).__name__} — "
            f"read-only. Writable fields: {writable}"
        )
        raise AttributeError(msg)

BeforeToolCallEvent

BeforeToolCallEvent(tool_name: str, tool_call_id: str, arguments: dict[str, Any])

Bases: ProtectedEvent

Event fired before each tool execution.

Writable fields

arguments: Modify tool arguments. cancel: Set True (or a string reason) to skip this tool call.

Read-only fields

tool_name: Name of the tool being called. tool_call_id: ID of the tool call.

Example

async def on_before_tool_call(self, event): if event.tool_name == "delete_file": event.cancel = "Blocked by security policy"

Source code in .sdk/src/tulip/hooks/provider.py

def __init__(self, tool_name: str, tool_call_id: str, arguments: dict[str, Any]) -> None:
    self._init("tool_name", tool_name)
    self._init("tool_call_id", tool_call_id)
    self._init("arguments", arguments)
    self._init("cancel", False)

__setattr__

__setattr__(name: str, value: Any) -> None

Only allow setting writable fields.

Source code in .sdk/src/tulip/hooks/provider.py

def __setattr__(self, name: str, value: Any) -> None:
    """Only allow setting writable fields."""
    if name.startswith("_") or name in self._writable:
        object.__setattr__(self, name, value)
    else:
        writable = ", ".join(sorted(self._writable)) or "none"
        msg = (
            f"Cannot set '{name}' on {type(self).__name__} — "
            f"read-only. Writable fields: {writable}"
        )
        raise AttributeError(msg)

AfterToolCallEvent

AfterToolCallEvent(tool_name: str, result: Any, error: str | None, *, tool_call_id: str = '', arguments: dict[str, Any] | None = None)

Bases: ProtectedEvent

Event fired after each tool execution.

Writable fields

retry: Set True to discard result and re-execute the tool. result: Replace the tool result.

Read-only fields

tool_name: Name of the tool that was called. tool_call_id: ID correlating this event with the matching BeforeToolCallEvent. Empty string if not supplied by the caller (e.g. tests constructing the event directly). arguments: The arguments the tool was invoked with (after any BeforeToolCallEvent mutations). Empty dict if not supplied. error: Error message (if failed).

Example

async def on_after_tool_call(self, event): # Mirror every tool call into an action queue keyed by id. self._queue.append({ "id": event.tool_call_id, "tool": event.tool_name, "args": event.arguments, "result": event.result, })

Source code in .sdk/src/tulip/hooks/provider.py

def __init__(
    self,
    tool_name: str,
    result: Any,
    error: str | None,
    *,
    tool_call_id: str = "",
    arguments: dict[str, Any] | None = None,
) -> None:
    self._init("tool_name", tool_name)
    self._init("tool_call_id", tool_call_id)
    self._init("arguments", arguments if arguments is not None else {})
    self._init("result", result)
    self._init("error", error)
    self._init("retry", False)

__setattr__

__setattr__(name: str, value: Any) -> None

Only allow setting writable fields.

Source code in .sdk/src/tulip/hooks/provider.py

def __setattr__(self, name: str, value: Any) -> None:
    """Only allow setting writable fields."""
    if name.startswith("_") or name in self._writable:
        object.__setattr__(self, name, value)
    else:
        writable = ", ".join(sorted(self._writable)) or "none"
        msg = (
            f"Cannot set '{name}' on {type(self).__name__} — "
            f"read-only. Writable fields: {writable}"
        )
        raise AttributeError(msg)

Lifecycle events

Read-only notifications fired at agent / iteration boundaries.

HookEvent

Bases: TulipEvent

Base class for hook lifecycle events.

HookResult

HookResult(provider_name: str, success: bool, result: Any = None, error: str | None = None)

Container for hook execution results.

Used to capture results and errors from hook provider execution.

Initialize hook result.

Parameters:

Name	Type	Description	Default
provider_name	str	Name of the hook provider	required
success	bool	Whether the hook executed successfully	required
result	Any	Return value from the hook (if any)	None
error	str | None	Error message (if failed)	None

Source code in .sdk/src/tulip/hooks/events.py

def __init__(
    self,
    provider_name: str,
    success: bool,
    result: Any = None,
    error: str | None = None,
) -> None:
    """Initialize hook result.

    Args:
        provider_name: Name of the hook provider
        success: Whether the hook executed successfully
        result: Return value from the hook (if any)
        error: Error message (if failed)
    """
    self.provider_name = provider_name
    self.success = success
    self.result = result
    self.error = error

__repr__

__repr__() -> str

Return string representation.

Source code in .sdk/src/tulip/hooks/events.py

def __repr__(self) -> str:
    """Return string representation."""
    status = "success" if self.success else f"error: {self.error}"
    return f"HookResult(provider={self.provider_name!r}, {status})"

BeforeInvocationEvent

Bases: HookEvent

Fired before agent invocation starts.

AfterInvocationEvent

Bases: HookEvent

Fired after agent invocation completes.

IterationStartEvent

Bases: HookEvent

Fired at the start of an agent iteration.

IterationEndEvent

Bases: HookEvent

Fired at the end of an agent iteration.

Built-in hooks

LoggingHook

LoggingHook(level: int = logging.INFO, logger_name: str = 'tulip.agent', extra: dict[str, Any] | None = None, log_arguments: bool = False, log_results: bool = False, priority: int = HookPriority.OBSERVABILITY_DEFAULT)

Bases: HookProvider

Hook provider that logs all lifecycle events.

Provides structured logging for agent execution with configurable log levels and optional extra context.

Example

Basic usage

registry.add_provider(LoggingHook())

With custom log level

registry.add_provider(LoggingHook(level=logging.DEBUG))

With structured context

registry.add_provider(LoggingHook( extra={"environment": "production", "service": "my-agent"} ))

Initialize logging hook.

Parameters:

Name	Type	Description	Default
level	int	Logging level (default: INFO)	INFO
logger_name	str	Name for the logger	'tulip.agent'
extra	dict[str, Any] | None	Extra context to include in all log records	None
log_arguments	bool	Whether to log tool arguments (may contain sensitive data)	False
log_results	bool	Whether to log tool results (may be verbose)	False
priority	int	Hook priority	OBSERVABILITY_DEFAULT

Source code in .sdk/src/tulip/hooks/builtin/logging.py

def __init__(
    self,
    level: int = logging.INFO,
    logger_name: str = "tulip.agent",
    extra: dict[str, Any] | None = None,
    log_arguments: bool = False,
    log_results: bool = False,
    priority: int = HookPriority.OBSERVABILITY_DEFAULT,
) -> None:
    """Initialize logging hook.

    Args:
        level: Logging level (default: INFO)
        logger_name: Name for the logger
        extra: Extra context to include in all log records
        log_arguments: Whether to log tool arguments (may contain sensitive data)
        log_results: Whether to log tool results (may be verbose)
        priority: Hook priority
    """
    self._level = level
    self._logger = logging.getLogger(logger_name)
    self._extra = extra or {}
    self._log_arguments = log_arguments
    self._log_results = log_results
    self._priority = priority

priority property

priority: int

Return hook priority.

name property

name: str

Return hook name.

on_before_invocation async

on_before_invocation(prompt: str, state: AgentState) -> AgentState

Log invocation start.

Parameters:

Name	Type	Description	Default
prompt	str	User prompt	required
state	AgentState	Agent state	required

Returns:

Type	Description
AgentState	Unchanged state

Source code in .sdk/src/tulip/hooks/builtin/logging.py

async def on_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Log invocation start.

    Args:
        prompt: User prompt
        state: Agent state

    Returns:
        Unchanged state
    """
    prompt_preview = prompt[:100] + "..." if len(prompt) > 100 else prompt
    self._log(
        "Agent invocation starting",
        run_id=state.run_id,
        agent_id=state.agent_id,
        prompt_preview=prompt_preview,
        prompt_length=len(prompt),
    )
    return state

on_after_invocation async

on_after_invocation(state: AgentState, success: bool) -> None

Log invocation completion.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution succeeded	required

Source code in .sdk/src/tulip/hooks/builtin/logging.py

async def on_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """Log invocation completion.

    Args:
        state: Final agent state
        success: Whether execution succeeded
    """
    duration_ms = (state.updated_at - state.started_at).total_seconds() * 1000
    self._log(
        "Agent invocation completed",
        run_id=state.run_id,
        agent_id=state.agent_id,
        success=success,
        iterations=state.iteration,
        confidence=state.confidence,
        tool_calls=len(state.tool_executions),
        errors=len(state.errors),
        duration_ms=duration_ms,
    )

on_before_tool_call async

on_before_tool_call(event: BeforeToolCallEvent) -> None

Log tool call start.

Parameters:

Name	Type	Description	Default
event	BeforeToolCallEvent	Write-protected event carrying tool_name and arguments. The hook only inspects them.	required

Source code in .sdk/src/tulip/hooks/builtin/logging.py

async def on_before_tool_call(self, event: BeforeToolCallEvent) -> None:
    """Log tool call start.

    Args:
        event: Write-protected event carrying ``tool_name`` and
            ``arguments``. The hook only inspects them.
    """
    log_data: dict[str, Any] = {"tool_name": event.tool_name}
    if self._log_arguments:
        log_data["arguments"] = event.arguments
    else:
        log_data["argument_keys"] = list(event.arguments.keys())

    self._log("Tool call starting", **log_data)

on_after_tool_call async

on_after_tool_call(event: AfterToolCallEvent) -> None

Log tool call completion.

Parameters:

Name	Type	Description	Default
event	AfterToolCallEvent	Write-protected event carrying tool_name, result, and error. The hook only inspects them.	required

Source code in .sdk/src/tulip/hooks/builtin/logging.py

async def on_after_tool_call(self, event: AfterToolCallEvent) -> None:
    """Log tool call completion.

    Args:
        event: Write-protected event carrying ``tool_name``,
            ``result``, and ``error``. The hook only inspects them.
    """
    log_data: dict[str, Any] = {
        "tool_name": event.tool_name,
        "success": event.error is None,
    }

    if event.error:
        log_data["error"] = event.error
    elif self._log_results and event.result is not None:
        result_str = str(event.result)
        log_data["result_preview"] = (
            result_str[:200] + "..." if len(result_str) > 200 else result_str
        )
        log_data["result_length"] = len(result_str)

    self._log("Tool call completed", **log_data)

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Log iteration start.

Parameters:

Name	Type	Description	Default
iteration	int	Iteration number	required
state	AgentState	Current state	required

Source code in .sdk/src/tulip/hooks/builtin/logging.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Log iteration start.

    Args:
        iteration: Iteration number
        state: Current state
    """
    self._log(
        "Iteration starting",
        run_id=state.run_id,
        iteration=iteration,
        max_iterations=state.max_iterations,
        confidence=state.confidence,
    )

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

Log iteration end.

Parameters:

Name	Type	Description	Default
iteration	int	Iteration number	required
state	AgentState	Current state	required

Source code in .sdk/src/tulip/hooks/builtin/logging.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Log iteration end.

    Args:
        iteration: Iteration number
        state: Current state
    """
    self._log(
        "Iteration completed",
        run_id=state.run_id,
        iteration=iteration,
        confidence=state.confidence,
        messages=len(state.messages),
    )

on_before_model_call async

on_before_model_call(event: BeforeModelCallEvent) -> None

Called before each model.complete() call.

Modify event.messages to change what the model sees. event.tools is read-only (inspect only).

Parameters:

Name	Type	Description	Default
event	BeforeModelCallEvent	Write-protected event. Writable: messages.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_before_model_call(
    self,
    event: BeforeModelCallEvent,
) -> None:
    """Called before each model.complete() call.

    Modify event.messages to change what the model sees.
    event.tools is read-only (inspect only).

    Args:
        event: Write-protected event. Writable: messages.
    """

on_after_model_call async

on_after_model_call(event: AfterModelCallEvent) -> None

Called after each model.complete() call.

Set event.retry = True to discard response and re-call. Set event.response to replace the response. event.messages is read-only.

Parameters:

Name	Type	Description	Default
event	AfterModelCallEvent	Write-protected event. Writable: response, retry.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_model_call(
    self,
    event: AfterModelCallEvent,
) -> None:
    """Called after each model.complete() call.

    Set event.retry = True to discard response and re-call.
    Set event.response to replace the response.
    event.messages is read-only.

    Args:
        event: Write-protected event. Writable: response, retry.
    """

register_hooks

register_hooks() -> dict[str, bool]

Return which hooks this provider implements.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping hook names to whether they are implemented.
dict[str, bool]	Useful for optimization - registry can skip calling unimplemented hooks.

Source code in .sdk/src/tulip/hooks/provider.py

def register_hooks(self) -> dict[str, bool]:
    """Return which hooks this provider implements.

    Returns:
        Dictionary mapping hook names to whether they are implemented.
        Useful for optimization - registry can skip calling unimplemented hooks.
    """
    return {
        "on_before_invocation": True,
        "on_after_invocation": True,
        "on_before_tool_call": True,
        "on_after_tool_call": True,
        "on_iteration_start": True,
        "on_iteration_end": True,
        "on_before_model_call": True,
        "on_after_model_call": True,
    }

StructuredLoggingHook

StructuredLoggingHook(level: int = logging.INFO, logger_name: str = 'tulip.agent.structured', extra: dict[str, Any] | None = None, include_timestamps: bool = True, priority: int = HookPriority.OBSERVABILITY_DEFAULT)

Bases: LoggingHook

Logging hook with JSON-structured output.

Extends LoggingHook to emit structured JSON logs suitable for log aggregation systems like ELK, Datadog, or CloudWatch.

Example

import json import logging

Configure JSON handler

handler = logging.StreamHandler() handler.setFormatter(JsonFormatter()) logging.getLogger("tulip.agent").addHandler(handler)

registry.add_provider(StructuredLoggingHook())

Initialize structured logging hook.

Parameters:

Name	Type	Description	Default
level	int	Logging level	INFO
logger_name	str	Logger name	'tulip.agent.structured'
extra	dict[str, Any] | None	Extra context for all logs	None
include_timestamps	bool	Whether to include ISO timestamps	True
priority	int	Hook priority	OBSERVABILITY_DEFAULT

Source code in .sdk/src/tulip/hooks/builtin/logging.py

def __init__(
    self,
    level: int = logging.INFO,
    logger_name: str = "tulip.agent.structured",
    extra: dict[str, Any] | None = None,
    include_timestamps: bool = True,
    priority: int = HookPriority.OBSERVABILITY_DEFAULT,
) -> None:
    """Initialize structured logging hook.

    Args:
        level: Logging level
        logger_name: Logger name
        extra: Extra context for all logs
        include_timestamps: Whether to include ISO timestamps
        priority: Hook priority
    """
    super().__init__(
        level=level,
        logger_name=logger_name,
        extra=extra,
        log_arguments=False,
        log_results=False,
        priority=priority,
    )
    self._include_timestamps = include_timestamps

priority property

priority: int

Return hook priority.

name property

name: str

Return hook name.

on_before_invocation async

on_before_invocation(prompt: str, state: AgentState) -> AgentState

Log invocation start.

Parameters:

Name	Type	Description	Default
prompt	str	User prompt	required
state	AgentState	Agent state	required

Returns:

Type	Description
AgentState	Unchanged state

Source code in .sdk/src/tulip/hooks/builtin/logging.py

async def on_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Log invocation start.

    Args:
        prompt: User prompt
        state: Agent state

    Returns:
        Unchanged state
    """
    prompt_preview = prompt[:100] + "..." if len(prompt) > 100 else prompt
    self._log(
        "Agent invocation starting",
        run_id=state.run_id,
        agent_id=state.agent_id,
        prompt_preview=prompt_preview,
        prompt_length=len(prompt),
    )
    return state

on_after_invocation async

on_after_invocation(state: AgentState, success: bool) -> None

Log invocation completion.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution succeeded	required

Source code in .sdk/src/tulip/hooks/builtin/logging.py

async def on_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """Log invocation completion.

    Args:
        state: Final agent state
        success: Whether execution succeeded
    """
    duration_ms = (state.updated_at - state.started_at).total_seconds() * 1000
    self._log(
        "Agent invocation completed",
        run_id=state.run_id,
        agent_id=state.agent_id,
        success=success,
        iterations=state.iteration,
        confidence=state.confidence,
        tool_calls=len(state.tool_executions),
        errors=len(state.errors),
        duration_ms=duration_ms,
    )

on_before_tool_call async

on_before_tool_call(event: BeforeToolCallEvent) -> None

Log tool call start.

Parameters:

Name	Type	Description	Default
event	BeforeToolCallEvent	Write-protected event carrying tool_name and arguments. The hook only inspects them.	required

Source code in .sdk/src/tulip/hooks/builtin/logging.py

async def on_before_tool_call(self, event: BeforeToolCallEvent) -> None:
    """Log tool call start.

    Args:
        event: Write-protected event carrying ``tool_name`` and
            ``arguments``. The hook only inspects them.
    """
    log_data: dict[str, Any] = {"tool_name": event.tool_name}
    if self._log_arguments:
        log_data["arguments"] = event.arguments
    else:
        log_data["argument_keys"] = list(event.arguments.keys())

    self._log("Tool call starting", **log_data)

on_after_tool_call async

on_after_tool_call(event: AfterToolCallEvent) -> None

Log tool call completion.

Parameters:

Name	Type	Description	Default
event	AfterToolCallEvent	Write-protected event carrying tool_name, result, and error. The hook only inspects them.	required

Source code in .sdk/src/tulip/hooks/builtin/logging.py

async def on_after_tool_call(self, event: AfterToolCallEvent) -> None:
    """Log tool call completion.

    Args:
        event: Write-protected event carrying ``tool_name``,
            ``result``, and ``error``. The hook only inspects them.
    """
    log_data: dict[str, Any] = {
        "tool_name": event.tool_name,
        "success": event.error is None,
    }

    if event.error:
        log_data["error"] = event.error
    elif self._log_results and event.result is not None:
        result_str = str(event.result)
        log_data["result_preview"] = (
            result_str[:200] + "..." if len(result_str) > 200 else result_str
        )
        log_data["result_length"] = len(result_str)

    self._log("Tool call completed", **log_data)

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Log iteration start.

Parameters:

Name	Type	Description	Default
iteration	int	Iteration number	required
state	AgentState	Current state	required

Source code in .sdk/src/tulip/hooks/builtin/logging.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Log iteration start.

    Args:
        iteration: Iteration number
        state: Current state
    """
    self._log(
        "Iteration starting",
        run_id=state.run_id,
        iteration=iteration,
        max_iterations=state.max_iterations,
        confidence=state.confidence,
    )

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

Log iteration end.

Parameters:

Name	Type	Description	Default
iteration	int	Iteration number	required
state	AgentState	Current state	required

Source code in .sdk/src/tulip/hooks/builtin/logging.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Log iteration end.

    Args:
        iteration: Iteration number
        state: Current state
    """
    self._log(
        "Iteration completed",
        run_id=state.run_id,
        iteration=iteration,
        confidence=state.confidence,
        messages=len(state.messages),
    )

on_before_model_call async

on_before_model_call(event: BeforeModelCallEvent) -> None

Called before each model.complete() call.

Modify event.messages to change what the model sees. event.tools is read-only (inspect only).

Parameters:

Name	Type	Description	Default
event	BeforeModelCallEvent	Write-protected event. Writable: messages.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_before_model_call(
    self,
    event: BeforeModelCallEvent,
) -> None:
    """Called before each model.complete() call.

    Modify event.messages to change what the model sees.
    event.tools is read-only (inspect only).

    Args:
        event: Write-protected event. Writable: messages.
    """

on_after_model_call async

on_after_model_call(event: AfterModelCallEvent) -> None

Called after each model.complete() call.

Set event.retry = True to discard response and re-call. Set event.response to replace the response. event.messages is read-only.

Parameters:

Name	Type	Description	Default
event	AfterModelCallEvent	Write-protected event. Writable: response, retry.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_model_call(
    self,
    event: AfterModelCallEvent,
) -> None:
    """Called after each model.complete() call.

    Set event.retry = True to discard response and re-call.
    Set event.response to replace the response.
    event.messages is read-only.

    Args:
        event: Write-protected event. Writable: response, retry.
    """

register_hooks

register_hooks() -> dict[str, bool]

Return which hooks this provider implements.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping hook names to whether they are implemented.
dict[str, bool]	Useful for optimization - registry can skip calling unimplemented hooks.

Source code in .sdk/src/tulip/hooks/provider.py

def register_hooks(self) -> dict[str, bool]:
    """Return which hooks this provider implements.

    Returns:
        Dictionary mapping hook names to whether they are implemented.
        Useful for optimization - registry can skip calling unimplemented hooks.
    """
    return {
        "on_before_invocation": True,
        "on_after_invocation": True,
        "on_before_tool_call": True,
        "on_after_tool_call": True,
        "on_iteration_start": True,
        "on_iteration_end": True,
        "on_before_model_call": True,
        "on_after_model_call": True,
    }

TelemetryHook

TelemetryHook(service_name: str = 'tulip-agent', tracer_name: str = 'tulip.hooks.telemetry', meter_name: str = 'tulip.hooks.telemetry', record_arguments: bool = False, record_results: bool = False, priority: int = HookPriority.OBSERVABILITY_MIN + 10)

Bases: HookProvider

Hook provider for OpenTelemetry tracing and metrics.

Provides automatic instrumentation for: - Trace spans for agent invocations and iterations - Trace spans for tool calls - Metrics for invocation duration, tool call counts, etc.

Requires the telemetry extra: pip install tulip[telemetry]

Example

from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import ConsoleSpanExporter, BatchSpanProcessor

Configure OpenTelemetry

provider = TracerProvider() provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter())) trace.set_tracer_provider(provider)

Add telemetry hook

registry.add_provider(TelemetryHook())

Initialize telemetry hook.

Parameters:

Name	Type	Description	Default
service_name	str	Service name for telemetry	'tulip-agent'
tracer_name	str	Name for the OpenTelemetry tracer	'tulip.hooks.telemetry'
meter_name	str	Name for the OpenTelemetry meter	'tulip.hooks.telemetry'
record_arguments	bool	Whether to record tool arguments as span attributes	False
record_results	bool	Whether to record tool results as span attributes	False
priority	int	Hook priority (default: early in observability range)	OBSERVABILITY_MIN + 10

Raises:

Type	Description
ImportError	If OpenTelemetry is not installed

Source code in .sdk/src/tulip/hooks/builtin/telemetry.py

def __init__(
    self,
    service_name: str = "tulip-agent",
    tracer_name: str = "tulip.hooks.telemetry",
    meter_name: str = "tulip.hooks.telemetry",
    record_arguments: bool = False,
    record_results: bool = False,
    priority: int = HookPriority.OBSERVABILITY_MIN + 10,
) -> None:
    """Initialize telemetry hook.

    Args:
        service_name: Service name for telemetry
        tracer_name: Name for the OpenTelemetry tracer
        meter_name: Name for the OpenTelemetry meter
        record_arguments: Whether to record tool arguments as span attributes
        record_results: Whether to record tool results as span attributes
        priority: Hook priority (default: early in observability range)

    Raises:
        ImportError: If OpenTelemetry is not installed
    """
    if not OTEL_AVAILABLE:
        msg = "OpenTelemetry is not installed. Install with: pip install tulip[telemetry]"
        raise ImportError(msg)

    self._service_name = service_name
    self._tracer = trace.get_tracer(tracer_name)
    self._meter = metrics.get_meter(meter_name)
    self._record_arguments = record_arguments
    self._record_results = record_results
    self._priority = priority

    # Active spans tracking + OTel context tokens for proper parent-
    # child propagation. Without attaching each span to OTel's current
    # context, exporters see every span as a new trace root — Langfuse
    # and similar backends then render N separate traces instead of
    # one tree per invocation. See issue #244.
    self._invocation_span: Span | None = None
    self._invocation_token: Any = None
    self._iteration_spans: dict[int, Span] = {}
    self._iteration_tokens: dict[int, Any] = {}
    self._current_iteration: int | None = None
    self._tool_spans: dict[str, tuple[Span, float]] = {}

    # Metrics
    self._invocation_counter = self._meter.create_counter(
        "tulip.invocations",
        description="Number of agent invocations",
        unit="1",
    )
    self._invocation_duration = self._meter.create_histogram(
        "tulip.invocation.duration",
        description="Duration of agent invocations",
        unit="ms",
    )
    self._iteration_counter = self._meter.create_counter(
        "tulip.iterations",
        description="Number of agent iterations",
        unit="1",
    )
    self._tool_call_counter = self._meter.create_counter(
        "tulip.tool_calls",
        description="Number of tool calls",
        unit="1",
    )
    self._tool_call_duration = self._meter.create_histogram(
        "tulip.tool_call.duration",
        description="Duration of tool calls",
        unit="ms",
    )
    self._tool_error_counter = self._meter.create_counter(
        "tulip.tool_errors",
        description="Number of tool call errors",
        unit="1",
    )

priority property

priority: int

Return hook priority.

name property

name: str

Return hook name.

on_before_invocation async

on_before_invocation(prompt: str, state: AgentState) -> AgentState

Start invocation span and attach it as the current OTel context.

Parameters:

Name	Type	Description	Default
prompt	str	User prompt	required
state	AgentState	Agent state	required

Returns:

Type	Description
AgentState	Unchanged state

Source code in .sdk/src/tulip/hooks/builtin/telemetry.py

async def on_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Start invocation span and attach it as the current OTel context.

    Args:
        prompt: User prompt
        state: Agent state

    Returns:
        Unchanged state
    """
    self._invocation_span = self._tracer.start_span(
        "agent.invocation",
        attributes={
            "tulip.run_id": state.run_id,
            "tulip.agent_id": state.agent_id or "",
            "tulip.prompt_length": len(prompt),
            "tulip.max_iterations": state.max_iterations,
            "service.name": self._service_name,
        },
    )
    # Attach the span to OTel's current context so every span the
    # agent creates inside this invocation is automatically parented
    # to it. Without this, child spans become disconnected trace
    # roots (issue #244).
    ctx = trace.set_span_in_context(self._invocation_span)
    self._invocation_token = otel_context.attach(ctx)
    self._invocation_counter.add(1, {"agent_id": state.agent_id or "default"})
    return state

on_after_invocation async

on_after_invocation(state: AgentState, success: bool) -> None

End invocation span.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution succeeded	required

Source code in .sdk/src/tulip/hooks/builtin/telemetry.py

async def on_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """End invocation span.

    Args:
        state: Final agent state
        success: Whether execution succeeded
    """
    if self._invocation_span:
        duration_ms = (state.updated_at - state.started_at).total_seconds() * 1000

        self._invocation_span.set_attributes(
            {
                "tulip.success": success,
                "tulip.iterations": state.iteration,
                "tulip.confidence": state.confidence,
                "tulip.tool_calls": len(state.tool_executions),
                "tulip.errors": len(state.errors),
                "tulip.duration_ms": duration_ms,
            }
        )

        if success:
            self._invocation_span.set_status(Status(StatusCode.OK))
        else:
            self._invocation_span.set_status(
                Status(StatusCode.ERROR, "Agent invocation failed")
            )

        self._invocation_span.end()
        self._invocation_span = None

        # Detach the invocation context — pair to the attach in
        # `on_before_invocation`. Detaching restores whatever
        # context was active before tulip took over.
        if self._invocation_token is not None:
            otel_context.detach(self._invocation_token)
            self._invocation_token = None

        # Record duration metric
        self._invocation_duration.record(
            duration_ms,
            {
                "agent_id": state.agent_id or "default",
                "success": str(success),
            },
        )

on_before_tool_call async

on_before_tool_call(event: BeforeToolCallEvent) -> None

Start tool call span.

Parameters:

Name	Type	Description	Default
event	BeforeToolCallEvent	Write-protected event carrying tool_name and arguments. The hook only inspects them.	required

Source code in .sdk/src/tulip/hooks/builtin/telemetry.py

async def on_before_tool_call(self, event: BeforeToolCallEvent) -> None:
    """Start tool call span.

    Args:
        event: Write-protected event carrying ``tool_name`` and
            ``arguments``. The hook only inspects them.
    """
    tool_name = event.tool_name
    span_attrs: dict[str, Any] = {
        "tulip.tool_name": tool_name,
    }

    if self._record_arguments:
        # Sanitize arguments for span attributes
        for key, value in event.arguments.items():
            attr_key = f"tulip.tool.arg.{key}"
            try:
                span_attrs[attr_key] = str(value)[:1000]  # Limit length
            except Exception:  # noqa: BLE001 — arbitrary user values; fall back to placeholder
                span_attrs[attr_key] = "<non-serializable>"

    span = self._tracer.start_span(f"tool.{tool_name}", attributes=span_attrs)
    self._tool_spans[tool_name] = (span, time.perf_counter())

    self._tool_call_counter.add(1, {"tool_name": tool_name})

on_after_tool_call async

on_after_tool_call(event: AfterToolCallEvent) -> None

End tool call span.

Parameters:

Name	Type	Description	Default
event	AfterToolCallEvent	Write-protected event carrying tool_name, result, and error.	required

Source code in .sdk/src/tulip/hooks/builtin/telemetry.py

async def on_after_tool_call(self, event: AfterToolCallEvent) -> None:
    """End tool call span.

    Args:
        event: Write-protected event carrying ``tool_name``,
            ``result``, and ``error``.
    """
    tool_name = event.tool_name
    error = event.error
    result = event.result
    if tool_name in self._tool_spans:
        span, start_time = self._tool_spans.pop(tool_name)
        duration_ms = (time.perf_counter() - start_time) * 1000

        span.set_attribute("tulip.duration_ms", duration_ms)

        if error:
            span.set_status(Status(StatusCode.ERROR, error))
            span.set_attribute("tulip.error", error[:1000])
            self._tool_error_counter.add(1, {"tool_name": tool_name})
        else:
            span.set_status(Status(StatusCode.OK))
            if self._record_results and result is not None:
                result_str = str(result)
                span.set_attribute("tulip.result_preview", result_str[:500])

        span.end()

        self._tool_call_duration.record(
            duration_ms,
            {
                "tool_name": tool_name,
                "success": str(error is None),
            },
        )

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Start iteration span as a child of the invocation span.

Parameters:

Name	Type	Description	Default
iteration	int	Iteration number	required
state	AgentState	Current state	required

Source code in .sdk/src/tulip/hooks/builtin/telemetry.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Start iteration span as a child of the invocation span.

    Args:
        iteration: Iteration number
        state: Current state
    """
    span = self._tracer.start_span(
        f"agent.iteration.{iteration}",
        attributes={
            "tulip.iteration": iteration,
            "tulip.confidence": state.confidence,
            "tulip.messages": len(state.messages),
        },
    )
    # Push this iteration's context onto OTel's stack so any tool
    # spans created during the iteration become children of it
    # (which is itself a child of the invocation).
    ctx = trace.set_span_in_context(span)
    self._iteration_spans[iteration] = span
    self._iteration_tokens[iteration] = otel_context.attach(ctx)
    self._current_iteration = iteration
    self._iteration_counter.add(1, {"agent_id": state.agent_id or "default"})

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

End iteration span.

Parameters:

Name	Type	Description	Default
iteration	int	Iteration number	required
state	AgentState	Current state	required

Source code in .sdk/src/tulip/hooks/builtin/telemetry.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """End iteration span.

    Args:
        iteration: Iteration number
        state: Current state
    """
    if iteration in self._iteration_spans:
        span = self._iteration_spans.pop(iteration)
        span.set_attributes(
            {
                "tulip.confidence_after": state.confidence,
                "tulip.messages_after": len(state.messages),
            }
        )
        span.set_status(Status(StatusCode.OK))
        span.end()
        # Detach the iteration context — pair to attach in
        # `on_iteration_start`. After detach, subsequent spans
        # parent to the invocation again.
        token = self._iteration_tokens.pop(iteration, None)
        if token is not None:
            otel_context.detach(token)
        if self._current_iteration == iteration:
            self._current_iteration = None

on_before_model_call async

on_before_model_call(event: BeforeModelCallEvent) -> None

Called before each model.complete() call.

Modify event.messages to change what the model sees. event.tools is read-only (inspect only).

Parameters:

Name	Type	Description	Default
event	BeforeModelCallEvent	Write-protected event. Writable: messages.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_before_model_call(
    self,
    event: BeforeModelCallEvent,
) -> None:
    """Called before each model.complete() call.

    Modify event.messages to change what the model sees.
    event.tools is read-only (inspect only).

    Args:
        event: Write-protected event. Writable: messages.
    """

on_after_model_call async

on_after_model_call(event: AfterModelCallEvent) -> None

Called after each model.complete() call.

Set event.retry = True to discard response and re-call. Set event.response to replace the response. event.messages is read-only.

Parameters:

Name	Type	Description	Default
event	AfterModelCallEvent	Write-protected event. Writable: response, retry.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_model_call(
    self,
    event: AfterModelCallEvent,
) -> None:
    """Called after each model.complete() call.

    Set event.retry = True to discard response and re-call.
    Set event.response to replace the response.
    event.messages is read-only.

    Args:
        event: Write-protected event. Writable: response, retry.
    """

register_hooks

register_hooks() -> dict[str, bool]

Return which hooks this provider implements.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping hook names to whether they are implemented.
dict[str, bool]	Useful for optimization - registry can skip calling unimplemented hooks.

Source code in .sdk/src/tulip/hooks/provider.py

def register_hooks(self) -> dict[str, bool]:
    """Return which hooks this provider implements.

    Returns:
        Dictionary mapping hook names to whether they are implemented.
        Useful for optimization - registry can skip calling unimplemented hooks.
    """
    return {
        "on_before_invocation": True,
        "on_after_invocation": True,
        "on_before_tool_call": True,
        "on_after_tool_call": True,
        "on_iteration_start": True,
        "on_iteration_end": True,
        "on_before_model_call": True,
        "on_after_model_call": True,
    }

NoOpTelemetryHook

NoOpTelemetryHook(priority: int = HookPriority.OBSERVABILITY_MIN + 10)

Bases: HookProvider

No-op telemetry hook for when OpenTelemetry is not available.

This hook does nothing but can be used as a drop-in replacement for TelemetryHook when telemetry is disabled.

Initialize no-op hook.

Parameters:

Name	Type	Description	Default
priority	int	Hook priority	OBSERVABILITY_MIN + 10

Source code in .sdk/src/tulip/hooks/builtin/telemetry.py

def __init__(self, priority: int = HookPriority.OBSERVABILITY_MIN + 10) -> None:
    """Initialize no-op hook.

    Args:
        priority: Hook priority
    """
    self._priority = priority

priority property

priority: int

Return hook priority.

name property

name: str

Return hook name.

on_before_invocation async

on_before_invocation(prompt: str, state: AgentState) -> AgentState

Called before agent starts processing.

Parameters:

Name	Type	Description	Default
prompt	str	The user prompt being processed	required
state	AgentState	Current agent state	required

Returns:

Type	Description
AgentState	Potentially modified agent state

Source code in .sdk/src/tulip/hooks/provider.py

async def on_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Called before agent starts processing.

    Args:
        prompt: The user prompt being processed
        state: Current agent state

    Returns:
        Potentially modified agent state
    """
    return state

on_after_invocation async

on_after_invocation(state: AgentState, success: bool) -> None

Called after agent completes processing.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution completed successfully	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """Called after agent completes processing.

    Args:
        state: Final agent state
        success: Whether execution completed successfully
    """

on_before_tool_call async

on_before_tool_call(event: BeforeToolCallEvent) -> None

Called before tool execution.

Modify event.arguments to change tool inputs. Set event.cancel = True or a string reason to skip execution. event.tool_name and event.tool_call_id are read-only.

Parameters:

Name	Type	Description	Default
event	BeforeToolCallEvent	Write-protected event. Writable: arguments, cancel.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_before_tool_call(
    self,
    event: BeforeToolCallEvent,
) -> None:
    """Called before tool execution.

    Modify event.arguments to change tool inputs.
    Set event.cancel = True or a string reason to skip execution.
    event.tool_name and event.tool_call_id are read-only.

    Args:
        event: Write-protected event. Writable: arguments, cancel.
    """

on_after_tool_call async

on_after_tool_call(event: AfterToolCallEvent) -> None

Called after tool execution.

Set event.retry = True to re-execute the tool. Set event.result to replace the tool result. event.tool_name and event.error are read-only.

Parameters:

Name	Type	Description	Default
event	AfterToolCallEvent	Write-protected event. Writable: result, retry.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_tool_call(
    self,
    event: AfterToolCallEvent,
) -> None:
    """Called after tool execution.

    Set event.retry = True to re-execute the tool.
    Set event.result to replace the tool result.
    event.tool_name and event.error are read-only.

    Args:
        event: Write-protected event. Writable: result, retry.
    """

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Called at the start of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the start of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

Called at the end of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the end of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_before_model_call async

on_before_model_call(event: BeforeModelCallEvent) -> None

Called before each model.complete() call.

Modify event.messages to change what the model sees. event.tools is read-only (inspect only).

Parameters:

Name	Type	Description	Default
event	BeforeModelCallEvent	Write-protected event. Writable: messages.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_before_model_call(
    self,
    event: BeforeModelCallEvent,
) -> None:
    """Called before each model.complete() call.

    Modify event.messages to change what the model sees.
    event.tools is read-only (inspect only).

    Args:
        event: Write-protected event. Writable: messages.
    """

on_after_model_call async

on_after_model_call(event: AfterModelCallEvent) -> None

Called after each model.complete() call.

Set event.retry = True to discard response and re-call. Set event.response to replace the response. event.messages is read-only.

Parameters:

Name	Type	Description	Default
event	AfterModelCallEvent	Write-protected event. Writable: response, retry.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_model_call(
    self,
    event: AfterModelCallEvent,
) -> None:
    """Called after each model.complete() call.

    Set event.retry = True to discard response and re-call.
    Set event.response to replace the response.
    event.messages is read-only.

    Args:
        event: Write-protected event. Writable: response, retry.
    """

register_hooks

register_hooks() -> dict[str, bool]

Return which hooks this provider implements.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping hook names to whether they are implemented.
dict[str, bool]	Useful for optimization - registry can skip calling unimplemented hooks.

Source code in .sdk/src/tulip/hooks/provider.py

def register_hooks(self) -> dict[str, bool]:
    """Return which hooks this provider implements.

    Returns:
        Dictionary mapping hook names to whether they are implemented.
        Useful for optimization - registry can skip calling unimplemented hooks.
    """
    return {
        "on_before_invocation": True,
        "on_after_invocation": True,
        "on_before_tool_call": True,
        "on_after_tool_call": True,
        "on_iteration_start": True,
        "on_iteration_end": True,
        "on_before_model_call": True,
        "on_after_model_call": True,
    }

ModelRetryHook

ModelRetryHook(max_retries: int = 3, initial_delay: float = 1.0, max_delay: float = 30.0, backoff_factor: float = 2.0, retry_on_empty: bool = True, priority: int = HookPriority.DEFAULT)

Bases: HookProvider

Retry model calls on throttle/rate limit with exponential backoff.

Catches empty responses and rate limit indicators, sets event.retry=True to trigger automatic re-invocation with increasing delays.

Works with all providers (OpenAI, Anthropic).

Parameters:

Name	Type	Description	Default
max_retries	int	Maximum retry attempts per model call.	3
initial_delay	float	First retry delay in seconds.	1.0
max_delay	float	Maximum delay between retries.	30.0
backoff_factor	float	Multiplier for each subsequent delay.	2.0
retry_on_empty	bool	Retry when model returns empty content.	True

Source code in .sdk/src/tulip/hooks/builtin/retry.py

def __init__(
    self,
    max_retries: int = 3,
    initial_delay: float = 1.0,
    max_delay: float = 30.0,
    backoff_factor: float = 2.0,
    retry_on_empty: bool = True,
    priority: int = HookPriority.DEFAULT,
) -> None:
    self._max_retries = max_retries
    self._initial_delay = initial_delay
    self._max_delay = max_delay
    self._backoff_factor = backoff_factor
    self._retry_on_empty = retry_on_empty
    self._priority = priority
    self._attempt = 0
    self.retries_total = 0

on_before_model_call async

on_before_model_call(event: Any) -> None

Reset attempt counter before each new model call.

Source code in .sdk/src/tulip/hooks/builtin/retry.py

async def on_before_model_call(self, event: Any) -> None:
    """Reset attempt counter before each new model call."""
    self._attempt = 0

on_after_model_call async

on_after_model_call(event: Any) -> None

Check response and retry if needed.

Source code in .sdk/src/tulip/hooks/builtin/retry.py

async def on_after_model_call(self, event: Any) -> None:
    """Check response and retry if needed."""
    response = event.response
    content = response.message.content or ""
    has_tool_calls = bool(response.message.tool_calls)

    # Determine if we should retry
    should_retry = False

    if self._retry_on_empty and not content and not has_tool_calls:
        should_retry = True

    if not should_retry:
        # Successful response — reset
        self._attempt = 0
        return

    # Check retry budget
    if self._attempt >= self._max_retries:
        logger.warning(
            "ModelRetryHook: exhausted %d retries, accepting empty response",
            self._max_retries,
        )
        self._attempt = 0
        return

    # Calculate delay with exponential backoff
    delay = min(
        self._initial_delay * (self._backoff_factor**self._attempt),
        self._max_delay,
    )

    self._attempt += 1
    self.retries_total += 1

    logger.info(
        "ModelRetryHook: retry %d/%d after %.1fs delay (empty response)",
        self._attempt,
        self._max_retries,
        delay,
    )

    from tulip.observability.emit import EV_HOOK_MODEL_RETRY, emit  # noqa: PLC0415

    await emit(
        EV_HOOK_MODEL_RETRY,
        attempt=self._attempt,
        max_retries=self._max_retries,
        delay_seconds=delay,
        reason="empty_response",
    )

    await asyncio.sleep(delay)
    event.retry = True

on_before_invocation async

on_before_invocation(prompt: str, state: AgentState) -> AgentState

Called before agent starts processing.

Parameters:

Name	Type	Description	Default
prompt	str	The user prompt being processed	required
state	AgentState	Current agent state	required

Returns:

Type	Description
AgentState	Potentially modified agent state

Source code in .sdk/src/tulip/hooks/provider.py

async def on_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Called before agent starts processing.

    Args:
        prompt: The user prompt being processed
        state: Current agent state

    Returns:
        Potentially modified agent state
    """
    return state

on_after_invocation async

on_after_invocation(state: AgentState, success: bool) -> None

Called after agent completes processing.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution completed successfully	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """Called after agent completes processing.

    Args:
        state: Final agent state
        success: Whether execution completed successfully
    """

on_before_tool_call async

on_before_tool_call(event: BeforeToolCallEvent) -> None

Called before tool execution.

Modify event.arguments to change tool inputs. Set event.cancel = True or a string reason to skip execution. event.tool_name and event.tool_call_id are read-only.

Parameters:

Name	Type	Description	Default
event	BeforeToolCallEvent	Write-protected event. Writable: arguments, cancel.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_before_tool_call(
    self,
    event: BeforeToolCallEvent,
) -> None:
    """Called before tool execution.

    Modify event.arguments to change tool inputs.
    Set event.cancel = True or a string reason to skip execution.
    event.tool_name and event.tool_call_id are read-only.

    Args:
        event: Write-protected event. Writable: arguments, cancel.
    """

on_after_tool_call async

on_after_tool_call(event: AfterToolCallEvent) -> None

Called after tool execution.

Set event.retry = True to re-execute the tool. Set event.result to replace the tool result. event.tool_name and event.error are read-only.

Parameters:

Name	Type	Description	Default
event	AfterToolCallEvent	Write-protected event. Writable: result, retry.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_tool_call(
    self,
    event: AfterToolCallEvent,
) -> None:
    """Called after tool execution.

    Set event.retry = True to re-execute the tool.
    Set event.result to replace the tool result.
    event.tool_name and event.error are read-only.

    Args:
        event: Write-protected event. Writable: result, retry.
    """

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Called at the start of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the start of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

Called at the end of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the end of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

register_hooks

register_hooks() -> dict[str, bool]

Return which hooks this provider implements.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping hook names to whether they are implemented.
dict[str, bool]	Useful for optimization - registry can skip calling unimplemented hooks.

Source code in .sdk/src/tulip/hooks/provider.py

def register_hooks(self) -> dict[str, bool]:
    """Return which hooks this provider implements.

    Returns:
        Dictionary mapping hook names to whether they are implemented.
        Useful for optimization - registry can skip calling unimplemented hooks.
    """
    return {
        "on_before_invocation": True,
        "on_after_invocation": True,
        "on_before_tool_call": True,
        "on_after_tool_call": True,
        "on_iteration_start": True,
        "on_iteration_end": True,
        "on_before_model_call": True,
        "on_after_model_call": True,
    }

GuardrailsHook

GuardrailsHook(config: GuardrailConfig | None = None, on_violation: Callable[[GuardrailViolation], None] | None = None, priority: int = HookPriority.SECURITY_DEFAULT)

Bases: HookProvider

Hook provider for security guardrails.

Provides: - Input validation and filtering - Output sanitization - PII detection and redaction - Dangerous content blocking - Tool allowlist/blocklist enforcement

Example

config = GuardrailConfig( block_dangerous_tools=frozenset({"shell", "exec"}), default_action=GuardrailAction.BLOCK, ) registry.add_provider(GuardrailsHook(config))

Initialize guardrails hook.

Parameters:

Name	Type	Description	Default
config	GuardrailConfig | None	Guardrail configuration	None
on_violation	Callable[[GuardrailViolation], None] | None	Callback for violations (receives GuardrailViolation)	None
priority	int	Hook priority (default: middle of security range)	SECURITY_DEFAULT

Source code in .sdk/src/tulip/hooks/builtin/guardrails.py

def __init__(
    self,
    config: GuardrailConfig | None = None,
    on_violation: Callable[[GuardrailViolation], None] | None = None,
    priority: int = HookPriority.SECURITY_DEFAULT,
) -> None:
    """Initialize guardrails hook.

    Args:
        config: Guardrail configuration
        on_violation: Callback for violations (receives GuardrailViolation)
        priority: Hook priority (default: middle of security range)
    """
    self._config = config or GuardrailConfig()
    self._on_violation = on_violation
    self._priority = priority
    self._violations: list[GuardrailViolation] = []

    # Compile patterns for efficiency
    self._compiled_pii: dict[str, re.Pattern[str]] = {
        name: re.compile(pattern) for name, pattern in self._config.pii_patterns.items()
    }
    self._compiled_blocked: dict[str, re.Pattern[str]] = {
        name: re.compile(pattern)
        for name, pattern in self._config.blocked_content_patterns.items()
    }

priority property

priority: int

Return hook priority.

name property

name: str

Return hook name.

violations property

violations: list[GuardrailViolation]

Get recorded violations.

clear_violations

clear_violations() -> None

Clear recorded violations.

Source code in .sdk/src/tulip/hooks/builtin/guardrails.py

def clear_violations(self) -> None:
    """Clear recorded violations."""
    self._violations.clear()

on_before_invocation async

on_before_invocation(prompt: str, state: AgentState) -> AgentState

Validate input prompt.

Parameters:

Name	Type	Description	Default
prompt	str	User prompt	required
state	AgentState	Agent state	required

Returns:

Type	Description
AgentState	State, potentially with metadata about violations

Raises:

Type	Description
ValueError	If prompt is blocked

Source code in .sdk/src/tulip/hooks/builtin/guardrails.py

async def on_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Validate input prompt.

    Args:
        prompt: User prompt
        state: Agent state

    Returns:
        State, potentially with metadata about violations

    Raises:
        ValueError: If prompt is blocked
    """
    violations: list[GuardrailViolation] = []

    # Check prompt length
    if len(prompt) > self._config.max_prompt_length:
        violation = GuardrailViolation(
            rule_name="max_prompt_length",
            description=f"Prompt exceeds maximum length ({len(prompt)} > {self._config.max_prompt_length})",
            action=self._get_action("max_prompt_length"),
            location="input",
        )
        violations.append(violation)
        self._record_violation(violation)

    # Check for blocked content
    violations.extend(self._check_blocked_content(prompt, "input"))

    # Check for PII
    pii_violations = self._check_pii(prompt, "input")
    violations.extend(pii_violations)

    # Handle blocking
    if self._should_block(violations):
        msg = f"Input blocked by guardrails: {violations[0].description}"
        raise ValueError(msg)

    # Store violations in metadata
    if violations:
        state = state.with_metadata(
            "guardrail_violations",
            [
                {
                    "rule_name": v.rule_name,
                    "description": v.description,
                    "action": v.action.value,
                    "location": v.location,
                }
                for v in violations
            ],
        )

    return state

on_before_tool_call async

on_before_tool_call(event: BeforeToolCallEvent) -> None

Validate tool call.

Parameters:

Name	Type	Description	Default
event	BeforeToolCallEvent	Write-protected event. The hook may mutate event.arguments (PII redaction) or set event.cancel to short-circuit a blocked tool.	required

Raises:

Type	Description
ValueError	If tool is blocked

Source code in .sdk/src/tulip/hooks/builtin/guardrails.py

async def on_before_tool_call(self, event: BeforeToolCallEvent) -> None:
    """Validate tool call.

    Args:
        event: Write-protected event. The hook may mutate
            ``event.arguments`` (PII redaction) or set
            ``event.cancel`` to short-circuit a blocked tool.

    Raises:
        ValueError: If tool is blocked
    """
    tool_name = event.tool_name
    arguments = event.arguments

    # Check tool blocklist
    if tool_name in self._config.block_dangerous_tools:
        violation = GuardrailViolation(
            rule_name="blocked_tool",
            description=f"Tool '{tool_name}' is blocked",
            action=GuardrailAction.BLOCK,
            location="tool_args",
        )
        self._record_violation(violation)
        msg = f"Tool '{tool_name}' is blocked by guardrails"
        raise ValueError(msg)

    # Check tool allowlist
    if (
        self._config.allow_only_tools is not None
        and tool_name not in self._config.allow_only_tools
    ):
        violation = GuardrailViolation(
            rule_name="tool_not_allowed",
            description=f"Tool '{tool_name}' is not in allowlist",
            action=GuardrailAction.BLOCK,
            location="tool_args",
        )
        self._record_violation(violation)
        msg = f"Tool '{tool_name}' is not allowed"
        raise ValueError(msg)

    # Check arguments for dangerous content
    args_str = str(arguments)
    violations = self._check_blocked_content(args_str, "tool_args")

    if self._should_block(violations):
        msg = f"Tool arguments blocked: {violations[0].description}"
        raise ValueError(msg)

    # Check for and optionally redact PII in arguments
    pii_violations = self._check_pii(args_str, "tool_args")
    if pii_violations and any(v.action == GuardrailAction.REDACT for v in pii_violations):
        # Redact PII from string arguments — write back to the event
        # so downstream hooks and the executor see the redacted form.
        redacted_args: dict[str, Any] = {}
        for key, value in arguments.items():
            if isinstance(value, str):
                redacted_args[key] = self._redact_pii(value)
            else:
                redacted_args[key] = value
        event.arguments = redacted_args

on_after_tool_call async

on_after_tool_call(event: AfterToolCallEvent) -> None

Validate tool result.

Parameters:

Name	Type	Description	Default
event	AfterToolCallEvent	Write-protected event carrying tool_name, result, and error.	required

Source code in .sdk/src/tulip/hooks/builtin/guardrails.py

async def on_after_tool_call(self, event: AfterToolCallEvent) -> None:
    """Validate tool result.

    Args:
        event: Write-protected event carrying ``tool_name``,
            ``result``, and ``error``.
    """
    result = event.result
    if result is None:
        return

    result_str = str(result)

    # Check result length
    if len(result_str) > self._config.max_tool_result_length:
        violation = GuardrailViolation(
            rule_name="max_tool_result_length",
            description=(
                f"Tool result exceeds maximum length "
                f"({len(result_str)} > {self._config.max_tool_result_length})"
            ),
            action=self._get_action("max_tool_result_length"),
            location="tool_result",
        )
        self._record_violation(violation)

    # Check for PII in results
    self._check_pii(result_str, "tool_result")

on_after_invocation async

on_after_invocation(state: AgentState, success: bool) -> None

Called after agent completes processing.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution completed successfully	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """Called after agent completes processing.

    Args:
        state: Final agent state
        success: Whether execution completed successfully
    """

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Called at the start of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the start of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

Called at the end of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the end of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_before_model_call async

on_before_model_call(event: BeforeModelCallEvent) -> None

Called before each model.complete() call.

Modify event.messages to change what the model sees. event.tools is read-only (inspect only).

Parameters:

Name	Type	Description	Default
event	BeforeModelCallEvent	Write-protected event. Writable: messages.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_before_model_call(
    self,
    event: BeforeModelCallEvent,
) -> None:
    """Called before each model.complete() call.

    Modify event.messages to change what the model sees.
    event.tools is read-only (inspect only).

    Args:
        event: Write-protected event. Writable: messages.
    """

on_after_model_call async

on_after_model_call(event: AfterModelCallEvent) -> None

Called after each model.complete() call.

Set event.retry = True to discard response and re-call. Set event.response to replace the response. event.messages is read-only.

Parameters:

Name	Type	Description	Default
event	AfterModelCallEvent	Write-protected event. Writable: response, retry.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_model_call(
    self,
    event: AfterModelCallEvent,
) -> None:
    """Called after each model.complete() call.

    Set event.retry = True to discard response and re-call.
    Set event.response to replace the response.
    event.messages is read-only.

    Args:
        event: Write-protected event. Writable: response, retry.
    """

register_hooks

register_hooks() -> dict[str, bool]

Return which hooks this provider implements.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping hook names to whether they are implemented.
dict[str, bool]	Useful for optimization - registry can skip calling unimplemented hooks.

Source code in .sdk/src/tulip/hooks/provider.py

def register_hooks(self) -> dict[str, bool]:
    """Return which hooks this provider implements.

    Returns:
        Dictionary mapping hook names to whether they are implemented.
        Useful for optimization - registry can skip calling unimplemented hooks.
    """
    return {
        "on_before_invocation": True,
        "on_after_invocation": True,
        "on_before_tool_call": True,
        "on_after_tool_call": True,
        "on_iteration_start": True,
        "on_iteration_end": True,
        "on_before_model_call": True,
        "on_after_model_call": True,
    }

SteeringHook

SteeringHook(model: Any, policy: str = '', evaluate_tools: bool = True, evaluate_responses: bool = False, interrupt_tools: set[str] | None = None, priority: int = HookPriority.SECURITY_DEFAULT)

Bases: HookProvider

LLM-powered steering for real-time agent guidance.

Evaluates each tool call before execution using a separate LLM. The steering model decides whether to proceed, guide (cancel with feedback), or interrupt (pause for human).

.. warning::

policy is advisory and can fail open. It is a second model asked to judge the first, so a judge that does not intervene is indistinguishable from one that approved. Measured against a self-hosted Qwen3.6-35B with policy="Never allow delete or destructive operations.": the judge did not intervene, and the agent reported "the users table has been successfully deleted."

Under the same model and the same prompt, a tool whose body calls :func:~tulip.control.admit held — the model still called the destructive tool, admission refused it, the side effect never ran, and the refusal landed on a chain that verifies.

Use this to steer an agent, not to stop it. For an action that must not happen, put the decision in code the model cannot argue with:

   from tulip.control import ControlPolicy, gate_tool

   agent = Agent(model=model, tools=[gate_tool(drop_table, policy=ControlPolicy())])

The two arguments below are not equally strong, and the difference is worth knowing before you rely on either.

Parameters:

Name	Type	Description	Default
model	Any	LLM for steering decisions (can be smaller/cheaper than main model).	required
policy	str	Natural language policy the agent must follow. Advisory — enforced by the judge model, and subject to the failure above.	''
evaluate_tools	bool	If True, evaluate tool calls before execution.	True
evaluate_responses	bool	If True, evaluate model responses after generation.	False
interrupt_tools	set[str] | None	Tools that always require human approval. Structural — a set-membership check that never consults the judge, so unlike policy it cannot fail open. Prefer it whenever the rule can be expressed as "this named tool, always".	None

Source code in .sdk/src/tulip/hooks/builtin/steering.py

def __init__(
    self,
    model: Any,
    policy: str = "",
    evaluate_tools: bool = True,
    evaluate_responses: bool = False,
    interrupt_tools: set[str] | None = None,
    priority: int = HookPriority.SECURITY_DEFAULT,
) -> None:
    self._model = model
    self._policy = policy
    self._evaluate_tools = evaluate_tools
    self._evaluate_responses = evaluate_responses
    self._interrupt_tools = interrupt_tools or set()
    self._priority = priority
    self._context = SteeringContext(policy=policy)
    self.decisions: list[SteeringDecision] = []

on_before_tool_call async

on_before_tool_call(event: Any) -> None

Evaluate tool call before execution.

Source code in .sdk/src/tulip/hooks/builtin/steering.py

async def on_before_tool_call(self, event: Any) -> None:
    """Evaluate tool call before execution."""
    if not self._evaluate_tools:
        return

    # Always interrupt for specified tools
    if event.tool_name in self._interrupt_tools:
        decision = SteeringDecision(
            action=SteeringAction.INTERRUPT,
            reason=f"Tool '{event.tool_name}' requires human approval",
        )
        self.decisions.append(decision)
        event.cancel = f"REQUIRES APPROVAL: {event.tool_name} needs human approval"
        return

    # LLM evaluation
    decision = await self._evaluate_tool_call(event.tool_name, event.arguments)
    self.decisions.append(decision)

    from tulip.observability.emit import EV_HOOK_STEERING_APPLIED, emit  # noqa: PLC0415

    if decision.action == SteeringAction.GUIDE:
        event.cancel = decision.guidance
        logger.info(
            "Steering GUIDE: %s(%s) — %s", event.tool_name, event.arguments, decision.reason
        )
        await emit(
            EV_HOOK_STEERING_APPLIED,
            action="guide",
            tool_name=event.tool_name,
            reason=decision.reason,
        )

    elif decision.action == SteeringAction.INTERRUPT:
        event.cancel = f"REQUIRES APPROVAL: {decision.reason}"
        logger.info("Steering INTERRUPT: %s — %s", event.tool_name, decision.reason)
        await emit(
            EV_HOOK_STEERING_APPLIED,
            action="interrupt",
            tool_name=event.tool_name,
            reason=decision.reason,
        )

    # Record in context
    self._context.record_tool_call(event.tool_name, event.arguments)

on_before_model_call async

on_before_model_call(event: Any) -> None

Track model calls in context.

Source code in .sdk/src/tulip/hooks/builtin/steering.py

async def on_before_model_call(self, event: Any) -> None:
    """Track model calls in context."""
    self._context.model_calls += 1

on_after_model_call async

on_after_model_call(event: Any) -> None

Optionally evaluate model responses.

Source code in .sdk/src/tulip/hooks/builtin/steering.py

async def on_after_model_call(self, event: Any) -> None:
    """Optionally evaluate model responses."""
    if not self._evaluate_responses:
        return

    content = event.response.message.content or ""
    if not content:
        return

    # Simple policy check — could be expanded to use steering LLM
    if self._policy and any(
        word in content.lower() for word in ["password", "secret", "credential"]
    ):
        logger.warning("Steering: response may contain sensitive info")

on_before_invocation async

on_before_invocation(prompt: str, state: AgentState) -> AgentState

Called before agent starts processing.

Parameters:

Name	Type	Description	Default
prompt	str	The user prompt being processed	required
state	AgentState	Current agent state	required

Returns:

Type	Description
AgentState	Potentially modified agent state

Source code in .sdk/src/tulip/hooks/provider.py

async def on_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Called before agent starts processing.

    Args:
        prompt: The user prompt being processed
        state: Current agent state

    Returns:
        Potentially modified agent state
    """
    return state

on_after_invocation async

on_after_invocation(state: AgentState, success: bool) -> None

Called after agent completes processing.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution completed successfully	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """Called after agent completes processing.

    Args:
        state: Final agent state
        success: Whether execution completed successfully
    """

on_after_tool_call async

on_after_tool_call(event: AfterToolCallEvent) -> None

Called after tool execution.

Set event.retry = True to re-execute the tool. Set event.result to replace the tool result. event.tool_name and event.error are read-only.

Parameters:

Name	Type	Description	Default
event	AfterToolCallEvent	Write-protected event. Writable: result, retry.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_tool_call(
    self,
    event: AfterToolCallEvent,
) -> None:
    """Called after tool execution.

    Set event.retry = True to re-execute the tool.
    Set event.result to replace the tool result.
    event.tool_name and event.error are read-only.

    Args:
        event: Write-protected event. Writable: result, retry.
    """

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Called at the start of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the start of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

Called at the end of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the end of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

register_hooks

register_hooks() -> dict[str, bool]

Return which hooks this provider implements.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping hook names to whether they are implemented.
dict[str, bool]	Useful for optimization - registry can skip calling unimplemented hooks.

Source code in .sdk/src/tulip/hooks/provider.py

def register_hooks(self) -> dict[str, bool]:
    """Return which hooks this provider implements.

    Returns:
        Dictionary mapping hook names to whether they are implemented.
        Useful for optimization - registry can skip calling unimplemented hooks.
    """
    return {
        "on_before_invocation": True,
        "on_after_invocation": True,
        "on_before_tool_call": True,
        "on_after_tool_call": True,
        "on_iteration_start": True,
        "on_iteration_end": True,
        "on_before_model_call": True,
        "on_after_model_call": True,
    }