Callback Manager

CallbackManager is the registry that runs your hooks around every model call, tool call, MCP call and skill call, and that runs input validators on incoming messages. You register callbacks per InvocationType, then pass the manager to graph.compile(callback_manager=...). For worked patterns see Use callbacks.

Import paths

All names below are exported from tenxgraph.utils. The validators are also available from tenxgraph.utils.validators.

# Callback system
from tenxgraph.utils import (
    CallbackManager,
    InvocationType,
    CallbackContext,
    BeforeInvokeCallback,
    AfterInvokeCallback,
    OnErrorCallback,
    BaseValidator,
)

# Validators
from tenxgraph.utils.validators import (
    PromptInjectionValidator,
    MessageContentValidator,
    ValidationError,
    register_default_validators,
)

CallbackManager

CallbackManager() takes no arguments and starts with empty registries. Callbacks for the same type run in registration order, each receiving the output of the previous one.

from tenxgraph.utils import CallbackManager

cbm = CallbackManager()
# app = graph.compile(callback_manager=cbm)

Registration methods

Each callback may be an instance of the matching abstract class or a plain function (sync or async) with the same arguments.

Execution methods

The framework calls these for you. You only call them directly in tests.

If a before or after callback raises, the manager runs the error callbacks for that type and then re-raises the exception. If an error callback itself raises, the failure is logged and the next error callback runs.

InvocationType

InvocationType is a string enum that selects which kind of call a callback fires on.

CallbackContext

CallbackContext is a dataclass passed as the first argument to every callback.

BeforeInvokeCallback

BeforeInvokeCallback runs before the call and may modify its input. Implement async __call__(self, context, input_data) and return the input to use, which can be the same object. Raise an exception to stop the call.

from tenxgraph.utils import (
    BeforeInvokeCallback,
    CallbackContext,
    CallbackManager,
    InvocationType,
)


class BlockEmptyInput(BeforeInvokeCallback):
    """Reject a model call that has no messages."""

    async def __call__(self, context: CallbackContext, input_data):
        if not input_data:
            raise ValueError(f"Empty input at node {context.node_name}")
        return input_data  # always return the input, modified or not


cbm = CallbackManager()
cbm.register_before_invoke(InvocationType.AI, BlockEmptyInput())

Returns the (possibly modified) input. Raises any exception to abort the invocation; the error callbacks run first.

AfterInvokeCallback

AfterInvokeCallback runs after a successful call and may modify its output. Implement async __call__(self, context, input_data, output_data) and return the output to use.

from tenxgraph.utils import (
    AfterInvokeCallback,
    CallbackContext,
    CallbackManager,
    InvocationType,
)


class AuditLogger(AfterInvokeCallback):
    """Log every completed model call."""

    async def __call__(self, context: CallbackContext, input_data, output_data):
        print(f"[audit] node={context.node_name} type={context.invocation_type.value}")
        return output_data  # keep the same type the framework expects


cbm = CallbackManager()
cbm.register_after_invoke(InvocationType.AI, AuditLogger())

Returns the (possibly modified) output. Return the same type you received, or the graph may fail downstream.

OnErrorCallback

OnErrorCallback runs when an invocation fails. Implement async __call__(self, context, input_data, error). Return a Message to offer a recovery value, or None to leave the error unhandled. Any other return value is ignored with a warning.

from tenxgraph.core.state import Message
from tenxgraph.utils import (
    CallbackContext,
    CallbackManager,
    InvocationType,
    OnErrorCallback,
)


class FriendlyFallback(OnErrorCallback):
    """Substitute a plain message when a model call fails."""

    async def __call__(self, context: CallbackContext, input_data, error: Exception):
        print(f"Error in {context.node_name}: {error}")
        return Message.text_message("Sorry, something went wrong.", role="assistant")


cbm = CallbackManager()
cbm.register_on_error(InvocationType.AI, FriendlyFallback())

Returns Message | None. When several error callbacks run, the manager keeps the result of the last one.

BaseValidator

BaseValidator is the abstract base for message validators. A validator needs one method, async validate(self, messages: list[Message]) -> bool, which returns True when the messages are acceptable and raises (usually ValidationError) when they are not. Validators run on new input messages before the graph executes.

from tenxgraph.core.state import Message
from tenxgraph.utils import BaseValidator, CallbackManager
from tenxgraph.utils.validators import ValidationError


class TopicPolicyValidator(BaseValidator):
    """Reject messages that mention a blocked term."""

    async def validate(self, messages: list[Message]) -> bool:
        for msg in messages:
            if "competitor" in msg.text().lower():
                raise ValidationError(
                    "Off-topic content detected",
                    "topic_policy",
                    {"content": msg.text()[:100]},
                )
        return True


cbm = CallbackManager()
cbm.register_input_validator(TopicPolicyValidator())

ValidationError

ValidationError is the exception validators raise. Its constructor is ValidationError(message: str, violation_type: str, details: dict[str, Any] | None = None).

The human-readable message is available through str(error).

PromptInjectionValidator

PromptInjectionValidator checks message text against built-in patterns for prompt injection, jailbreak phrases, role manipulation, system prompt leakage, delimiter confusion and template injection (OWASP LLM01:2025). It also checks for oversized input, encoded payloads (base64, hex, heavy non-ASCII), several suspicious keywords in one message, and payload splitting markers.

from tenxgraph.utils import CallbackManager
from tenxgraph.utils.validators import PromptInjectionValidator

validator = PromptInjectionValidator(
    strict_mode=True,                        # raise ValidationError on a violation
    max_length=10000,                        # longest allowed message text
    blocked_patterns=[r"(?i)internal\s+only"],  # extra regex patterns
    suspicious_keywords=["exfiltrate"],      # extra keywords
)

cbm = CallbackManager()
cbm.register_input_validator(validator)

validate(messages) returns True or raises ValidationError. The built-in patterns are broad, so expect some false positives on legitimate text (for example words like “sudo” or very long tokens). Test with your real traffic before enabling strict mode in production.

MessageContentValidator

MessageContentValidator checks message structure: the role must be allowed, and list content may not exceed a block limit.

from tenxgraph.utils import CallbackManager
from tenxgraph.utils.validators import MessageContentValidator

validator = MessageContentValidator(
    allowed_roles=["user", "assistant", "system", "tool"],
    max_content_blocks=50,
)

cbm = CallbackManager()
cbm.register_input_validator(validator)

Raises ValidationError with violation_type "invalid_role" or "too_many_blocks".

register_default_validators

register_default_validators(callback_manager, strict_mode=True) registers a PromptInjectionValidator (with the given strict_mode) and a MessageContentValidator (with defaults) on the manager in one call.

from tenxgraph.utils import CallbackManager
from tenxgraph.utils.validators import register_default_validators

cbm = CallbackManager()
register_default_validators(cbm, strict_mode=True)
# app = graph.compile(callback_manager=cbm)

Common errors