Worker Callback¶
Introduction¶
The Worker Callback mechanism will invoke pre-defined hook methods at key points during worker execution, allowing you to add cross-cutting concerns such as logging, validation, monitoring, and error handling without modifying your business logic. This tutorial will guide you through understanding and using the Worker Callback mechanism.
WorkerCallback is the base class of all callback instances that are invokded around a worker. You can implement the following three methods to subclass WorkerCallback:
| Method | Description |
|---|---|
on_worker_start() | Called before worker execution. Use for input validation, logging, or monitoring |
on_worker_end() | Called after successful execution. Use for result logging, event publishing, or metrics |
on_worker_error() | Called when an exception is raised. Use for error handling, logging, or suppression |
Creating a Custom Callback¶
Step 1: Define a Class¶
To create a custom callback, simply subclass WorkerCallback and implement the methods above. You don't need to implement all three methods, but only implement the ones that you need. The base WorkerCallback class provides default implementations that do nothing. Here we define a LoggingCallback class that implements all of the three hook methods.
from typing import Any, Dict, Optional
from bridgic.core.automa import Automa
from bridgic.core.automa.worker import WorkerCallback
class LoggingCallback(WorkerCallback):
"""Log worker lifecycle events."""
def __init__(self, tag: str = None):
self._tag = tag or ""
async def on_worker_start(
self,
key: str,
is_top_level: bool = False,
parent: Optional[Automa] = None,
arguments: Dict[str, Any] = None,
) -> None:
print(self._tag + f"[START] {key} args={arguments}")
async def on_worker_end(
self,
key: str,
is_top_level: bool = False,
parent: Optional[Automa] = None,
arguments: Dict[str, Any] = None,
result: Any = None,
) -> None:
print(self._tag + f"[END] {key} result={result}")
async def on_worker_error(
self,
key: str,
is_top_level: bool = False,
parent: Optional[Automa] = None,
arguments: Dict[str, Any] = None,
error: Exception = None,
) -> bool:
print(self._tag + f"[ERROR] {key} -> {error}")
return False # Returning False means don't suppress the exception.
Step 2: Choose Building Mode¶
Callbacks are instantiated through WorkerCallbackBuilder, which delays construction until the worker is created and lets you control its sharing mode.
- Shared instance mode (
is_shared=True, default): all workers within the declaration scope reuse the same callback instance. This ideal for stateful integrations (e.g., a tracing client). - Independent instance mode (
is_shared=False): every worker receives its own callback instance. This is useful when you need isolated state or thread safety.
from bridgic.core.automa.worker import WorkerCallbackBuilder
# Build in shared instance mode.
shared_builder = WorkerCallbackBuilder(LoggingCallback)
isolated_builder = WorkerCallbackBuilder(LoggingCallback, is_shared=False)
# Build in independent instance mode.
shared_builder_with_args = WorkerCallbackBuilder(LoggingCallback, init_kwargs={"tag": "-"})
isolated_builder_with_args = WorkerCallbackBuilder(LoggingCallback, init_kwargs={"tag": "-"}, is_shared=False)
Step 3: Decide the Scope¶
Choose where the builder should take effect. Bridgic merges builders from the widest scope to the narrowest one:
| Level | Registration Method | Scope of Effect |
|---|---|---|
| Global-level | GlobalSetting | Applies to all workers across all Automa instances |
| Automa-level | RunningOptions | Applies to every worker inside one Automa instance |
| Worker-level | @worker or add_worker() | Applies only to the targeted worker |
Global-Level Configuration
The following example shows how to configure callbacks at the global level using GlobalSetting, which applies the callback to all workers across all Automa instances in your application.
from bridgic.core.automa import GraphAutoma, worker
from bridgic.core.automa.worker import WorkerCallbackBuilder
from bridgic.core.config import GlobalSetting
GlobalSetting.set(callback_builders=[WorkerCallbackBuilder(LoggingCallback)])
class MyAutoma(GraphAutoma):
@worker(is_start=True)
async def step1(self, x: int) -> int:
return x + 1
automa = MyAutoma(name="test-automa") # Will log for all workers inside.
await automa.arun(x=10)
[START] test-automa args={'args': (), 'kwargs': {'x': 10}, 'feedback_data': None}
[START] step1 args={'args': (), 'kwargs': {'x': 10}}
[END] step1 result=11
[END] test-automa result=None
Automa-Level Configuration
The following example shows how to configure callbacks at the automa level using RunningOptions, which applies the callback to all workers within a specific Automa instance.
from bridgic.core.automa import GraphAutoma, RunningOptions, worker
from bridgic.core.automa.worker import WorkerCallbackBuilder
# Because this example is under jupyter-notebook environment, GlobalSetting needs to be reset.
# In real deveopment, this line is not necessary to have a default global setting.
GlobalSetting.set(callback_builders=[])
class MyAutoma(GraphAutoma):
@worker(is_start=True)
async def step1(self, x: int) -> int:
return x + 1
running_options = RunningOptions(callback_builders=[WorkerCallbackBuilder(LoggingCallback)])
automa = MyAutoma(name="test-automa", running_options=running_options) # Will log for all workers inside.
await automa.arun(x=10)
[START] test-automa args={'args': (), 'kwargs': {'x': 10}, 'feedback_data': None}
[START] step1 args={'args': (), 'kwargs': {'x': 10}}
[END] step1 result=11
[END] test-automa result=None
Worker-Level Configuration
The following example shows how to configure callbacks at the worker level by passing callback_builders to the @worker decorator, which applies the callback only to the specific worker.
from bridgic.core.automa import GraphAutoma, worker
from bridgic.core.automa.worker import WorkerCallbackBuilder
# Because this example is under jupyter-notebook environment, GlobalSetting needs to be reset.
# In real deveopment, this line is not necessary to have a default global setting.
GlobalSetting.set(callback_builders=[])
class MyAutoma(GraphAutoma):
@worker(
is_start=True,
callback_builders=[WorkerCallbackBuilder(LoggingCallback)],
)
async def step1(self, x: int) -> int:
return x + 1
automa = MyAutoma(name="test-automa") # Will only log for "step1" worker.
await automa.arun(x=10)
[START] step1 args={'args': (), 'kwargs': {'x': 10}}
[END] step1 result=11
Features you Need to Know¶
Callback Propagation¶
Nested automa will inherit callbacks from their parent (even higher ancestor) scope by reading the initialized running options. This ensures instrumentation remains consistent across multi levels. When you set a callback in the RunningOptions during automa initialization, all workers in any nested automata will automatically inherit that callback:
In the following example, the LoggingCallback configured at the top-level automa propagates to:
- All workers directly in
TopAutoma(top_worker,nested_automa_worker) - All workers inside the nested
InnerAutoma(inner_worker)
from bridgic.core.automa import GraphAutoma, RunningOptions, worker
from bridgic.core.automa.worker import WorkerCallback, WorkerCallbackBuilder
from bridgic.core.config import GlobalSetting
# Top-level automa
class TopAutoma(GraphAutoma):
@worker(is_start=True)
async def top_worker(self, x: int) -> int:
return x + 1
# Inner automa (will be used as a nested worker)
class InnerAutoma(GraphAutoma):
@worker(is_start=True)
async def inner_worker(self, x: int) -> int:
return x * 2
# Configure callback at global setting, with <Global> tag.
GlobalSetting.set(
callback_builders=[
WorkerCallbackBuilder(LoggingCallback, init_kwargs={"tag": "<Global>"}),
]
)
# Configure callback at top-level automa, with <Automa> tag.
running_options = RunningOptions(
callback_builders=[
WorkerCallbackBuilder(LoggingCallback, init_kwargs={"tag": "<Automa>"})
]
)
automa = TopAutoma(name="top-automa", running_options=running_options)
# Add a instance of InnerAutoma as a worker.
automa.add_worker("nested_automa_as_worker", InnerAutoma(name="inner-automa"), dependencies=["top_worker"])
# When executed:
# - Callbacks that from GlobalSetting will be propagated to all workers application-wide.
# - Callbacks that from RunningOptions will be propagated to all workers inside the "top-level" automa.
await automa.arun(x=10)
<Global>[START] top-automa args={'args': (), 'kwargs': {'x': 10}, 'feedback_data': None}
<Automa>[START] top-automa args={'args': (), 'kwargs': {'x': 10}, 'feedback_data': None}
<Global>[START] top_worker args={'args': (), 'kwargs': {'x': 10}}
<Automa>[START] top_worker args={'args': (), 'kwargs': {'x': 10}}
<Global>[END] top_worker result=11
<Automa>[END] top_worker result=11
<Global>[START] nested_automa_as_worker args={'args': (11,), 'kwargs': {'feedback_data': None, 'x': 10}}
<Automa>[START] nested_automa_as_worker args={'args': (11,), 'kwargs': {'feedback_data': None, 'x': 10}}
<Global>[START] inner_worker args={'args': (11,), 'kwargs': {}}
<Automa>[START] inner_worker args={'args': (11,), 'kwargs': {}}
<Global>[END] inner_worker result=22
<Automa>[END] inner_worker result=22
<Global>[END] nested_automa_as_worker result=None
<Automa>[END] nested_automa_as_worker result=None
<Global>[END] top-automa result=None
<Automa>[END] top-automa result=None
Dynamical Topology Support¶
GraphAutoma.add_worker() and related APIs allow you to modify the topology at runtime. When a new worker is added, Bridgic automatically builds its callback list using the current global builders, the builders from its ancestors' running options, and the builders passed with the add_worker() call. As a result:
- Dynamically added workers receive the same instrumentation guarantees as statically declared ones.
- Nested automa inserted later as a new worker inherits callbacks from its ancestors scopes.
This design keeps long-running agentic systems observable even as they grow or reconfigure themselves during execution.
Exception Handling¶
Exception Type Matching¶
The on_worker_error() method allows for fine-grained and flexible error handling by inspecting the type annotation of its error parameter. You can indicate exactly which exception types you want your handler to respond to, by annotating the error parameter with a specific exception type. At runtime, the Bridgic framework will automatically match and invoke your callback only for those exceptions that match the annotation.
Below is a simple example comparison table:
Type annotation of error | The matched exception type(s) to trigger on_worker_error |
|---|---|
Exception | All exceptions |
ValueError | ValueError and its subclasses (e.g., UnicodeDecodeError) |
Union[Type1, Type2, ...] | Type1 and Type2 will be considered to be matched |
import warnings
from typing import Union
from bridgic.core.automa import Automa
from bridgic.core.automa.worker import WorkerCallback
class ValueErrorHandler(WorkerCallback):
async def on_worker_error(
self,
key: str,
is_top_level: bool = False,
parent: Optional[Automa] = None,
arguments: Dict[str, Any] = None,
error: ValueError = None
) -> bool:
warnings.warn("ValueError in %s: %s", key, error)
return True # Swallow ValueError.
class MultipleErrorHandler(WorkerCallback):
async def on_worker_error(
self,
key: str,
is_top_level: bool = False,
parent: Optional[Automa] = None,
arguments: Dict[str, Any] = None,
error: Union[KeyError, TypeError] = None
) -> bool:
warnings.warn("Recoverable issue: %s", error)
return False # Re-raise it.
Exception Suppression¶
The return value of on_worker_error will determine whether to suppress the captured exception:
| Value | Behavior |
|---|---|
True | Means to suppress the exception; the worker result becomes None. |
False | Means to observe only; the framework re-raises after all callbacks finish. |
Specially, to ensure human-in-the-loop flows stay intact, InteractionException should never be suppressed.
The framework calls every matching callback, so you can compose specialized handlers with broader "catch-all" callbacks.
Best Practices¶
Keep callbacks lightweight: Callback methods are called for each worker that they are responsible for, so keep them fast and avoid blocking operations.
Use appropriate scope: Use global-level for application-wide concerns, automa-level for specific instance, and worker-level for fine-grained control.
Handle exceptions carefully: Be thoughtful about which exceptions to suppress. Suppressing exceptions can hide bugs and make debugging difficult.
Use shared instances wisely: Shared instances are great for maintaining connections or state, but be aware of thread-safety concerns.
Leverage the parent parameter: The
parentparameter gives you access to the automa's context, allowing you to post events, request feedback, or interact with the automa's state.
Next Steps¶
- Learn about Observability to see how callbacks enable system transparency
- Explore Callback Integrations for ready-to-use callback implementations