10xGraphv0.9.2Search

    Callback Manager

    In shortCallbackManager, BeforeInvokeCallback, AfterInvokeCallback, OnErrorCallback, BaseValidator, PromptInjectionValidator — hook into every LLM, tool, and MCP

    • 4 min read
    • 14 sections
    • Updated
    • v0.9.2
    • Markdown

    When to use this

    Use the callback system to:

    • Block or modify inputs before the LLM/tool is called (input validation, prompt injection protection).
    • Inspect or modify outputs after the LLM/tool returns (response filtering, logging).
    • Handle errors gracefully and optionally return a recovery value.
    • Register structured validators as a simpler alternative to manual before-invoke callbacks.

    Pass a CallbackManager to graph.compile(callback_manager=...).

    Import paths

    Python
    from tenxgraph.utils import (
    CallbackManager,
    InvocationType,
    BeforeInvokeCallback,
    AfterInvokeCallback,
    OnErrorCallback,
    BaseValidator,
    CallbackContext,
)
from tenxgraph.utils.validators import (
    PromptInjectionValidator,
    MessageContentValidator,
    ValidationError,
    register_default_validators,
)

    CallbackManager

    Central registry and executor for all callbacks and validators.

    Python
    from tenxgraph.utils import CallbackManager

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

    Methods

    Method Signature Description
    register_before_invoke (invocation_type, callback) Register a callback fired before an invocation.
    register_after_invoke (invocation_type, callback) Register a callback fired after an invocation.
    register_on_error (invocation_type, callback) Register an error-handler callback.
    register_input_validator (validator: BaseValidator) Register a validator (simpler than a before-invoke callback).
    execute_before_invoke async (context, input_data) → input_data Called internally by the framework.
    execute_after_invoke async (context, input_data, output_data) → output_data Called internally by the framework.
    execute_on_error async (context, input_data, error) → Message | None Called internally by the framework.
    execute_validators async (messages: list[Message]) → bool Called internally; runs all registered validators.

    InvocationType

    Selects which invocation type a callback fires on.

    Value Fires when
    InvocationType.AI The LLM provider is called.
    InvocationType.TOOL A local Python tool function is called.
    InvocationType.MCP An MCP tool call is made.
    InvocationType.INPUT_VALIDATION Input validation phase (validators).
    InvocationType.SKILL The model calls a skill tool: activate_skill or read_skill_resource. function_name is the tool name. Skill tools fire SKILL instead of TOOL. Session-mode preloading is not a tool call and fires nothing.

    CallbackContext

    Passed to every callback with metadata about the current invocation.

    Field Type Description
    invocation_type InvocationType The type of invocation that fired this callback.
    node_name str Name of the graph node executing.
    function_name str | None Name of the tool or skill function (if applicable).
    metadata dict | None Additional context from the framework.

    BeforeInvokeCallback

    Fires before the LLM/tool call. Can modify or block the input.

    Python
    from tenxgraph.utils.callbacks import BeforeInvokeCallback, CallbackContext
from tenxgraph.utils import InvocationType

class UpperCaseInput(BeforeInvokeCallback):
    async def __call__(self, context: CallbackContext, input_data) -> ...:
        # input_data is the list of messages about to be sent to the LLM
        for msg in input_data:
            if hasattr(msg, "content") and isinstance(msg.content, str):
                msg.content = msg.content.upper()
        return input_data   # must return (possibly modified) input_data

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

    To block execution, raise any exception from the callback.

    AfterInvokeCallback

    Fires after the LLM/tool returns. Can modify the response.

    Python
    from tenxgraph.utils.callbacks import AfterInvokeCallback, CallbackContext

class AuditLogger(AfterInvokeCallback):
    async def __call__(self, context: CallbackContext, input_data, output_data):
        print(f"[AUDIT] node={context.node_name} type={context.invocation_type}")
        return output_data   # must return (possibly modified) output_data

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

    OnErrorCallback

    Fires when any exception occurs during an invocation. Can return a recovery Message or None (re-raise).

    Python
    from tenxgraph.utils.callbacks import OnErrorCallback, CallbackContext

class FallbackOnError(OnErrorCallback):
    async def __call__(self, context: CallbackContext, input_data, error: Exception):
        print(f"Error in {context.node_name}: {error}")
        return None   # return None to re-raise, or a Message to substitute

cbm.register_on_error(InvocationType.AI, FallbackOnError())

    BaseValidator and the validator API

    Validators focus specifically on message content validation. They are simpler than callbacks because they only need one method.

    Python
    from tenxgraph.utils import BaseValidator
from tenxgraph.utils.validators import ValidationError

class AllowedTopicsValidator(BaseValidator):
    async def validate(self, messages) -> 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.register_input_validator(AllowedTopicsValidator())

    PromptInjectionValidator

    Detects OWASP LLM01:2025 prompt injection attacks.

    Python
    from tenxgraph.utils.validators import PromptInjectionValidator

validator = PromptInjectionValidator(
    strict_mode=True,        # raises ValidationError (default)
    max_length=10000,        # max allowed message length
    blocked_patterns=[],     # additional regex patterns
    suspicious_keywords=[],  # additional keywords
)
cbm.register_input_validator(validator)

    Constructor parameters

    Parameter Type Default Description
    strict_mode bool True True raises ValidationError; False logs a warning and sanitizes.
    max_length int 10000 Maximum message content length in characters.
    blocked_patterns list[str] [] Additional regex patterns to block (merged with default patterns).
    suspicious_keywords list[str] [] Additional keywords to flag.

    MessageContentValidator

    Validates message structure integrity.

    Python
    from tenxgraph.utils.validators import MessageContentValidator

validator = MessageContentValidator(
    allowed_roles=["user", "assistant", "system"],
    max_content_blocks=50,
)
cbm.register_input_validator(validator)

    ValidationError

    Raised by validators when validation fails.

    Attribute Type Description
    violation_type str Category: "injection_pattern", "length_exceeded", "encoding_attack", "invalid_role", etc.
    details dict Extra context: matched pattern, content sample, input length.

    register_default_validators

    One-call setup that registers both PromptInjectionValidator (strict) and MessageContentValidator.

    Python
    from tenxgraph.utils.validators import register_default_validators

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

    Common errors

    Error Cause Fix
    Callbacks never fire callback_manager not passed to compile(). graph.compile(callback_manager=cbm).
    ValidationError brings down the server Not caught outside the graph. Wrap ainvoke/astream in try/except ValidationError.
    Before-invoke callback receives None as input_data LLM call with an empty message list. Check for empty messages in the callback before iterating.
    After-invoke callback changes return type, graph breaks Returning wrong type from __call__. Always return the same type as output_data.
    Last updated for v0.9.2Edit this page on GitHubReport an issue