Display Callbacks
Hook into agent output events for custom terminals, logs, and dashboards
from praisonaiagents import Agent
agent = Agent(name="assistant", instructions="Be helpful.", output=True)
agent.start("Summarise this log file.")Display callbacks let you react to agent events — tool calls, LLM turns, errors — without changing agent logic.
The user runs the agent; display callbacks fire on tool calls, LLM turns, and errors for custom terminals or dashboards.
How It Works
Quick Start
- Register a callback
python from praisonaiagents import Agent, register_display_callback def on_interaction(message=None, response=None, **kwargs): print(f"User: {message}") print(f"Agent: {response}") register_display_callback("interaction", on_interaction) agent = Agent(name="assistant", instructions="Be concise.") agent.start("Say hello in one sentence.") - Log tool calls
python from praisonaiagents import Agent, register_display_callback def on_tool_call(tool_name=None, tool_input=None, tool_output=None, elapsed_time=None, success=None, **kwargs): status = "ok" if success else "failed" print(f"[{status}] {tool_name}({tool_input}) -> {tool_output}") register_display_callback("tool_call", on_tool_call) agent = Agent( name="researcher", instructions="Use tools when helpful.", tools=["web_search"], ) agent.start("Latest Python release version?")Fires on non-streaming and streaming turns since PR #4735. See Streaming coverage.
- Async callback for dashboards
python import asyncio from praisonaiagents import Agent, register_display_callback async def push_to_ui(message=None, response=None, **kwargs): await asyncio.sleep(0) # replace with your WebSocket / HTTP call print({"message": message, "response": response}) register_display_callback("interaction", push_to_ui, is_async=True) agent = Agent(name="assistant", instructions="Be helpful.") agent.start("Summarise loop detection in one line.")
Callback Types
| Type | When it fires |
|---|---|
interaction | User message and agent response |
tool_call | Before or during a tool invocation |
error | Agent or tool error |
llm_start / llm_end | LLM request lifecycle — fires on both sync and async agents (since #4887) |
llm_content | Streaming token chunks — the primary interactive TUI consumes this to render live intermediate narrative during tool-using turns |
autonomy_iteration | Autonomous loop iteration |
autonomy_doom_loop | Doom loop detected in autonomy mode |
retry | Retry after transient failure |
Register with register_display_callback(type, fn, is_async=False). Alias: add_display_callback.
Streaming coverage
Since PR #4735, tool_call fires on all three execution paths — not just non-streaming.
| Path | SDK entry point | tool_call fires |
|---|---|---|
| Non-streaming | llm/openai_client.py | ✅ Always |
| Sync-OpenAI streaming | agent/chat_mixin.py::_start_stream_impl | ✅ Since #4735 |
| Custom-LLM streaming | llm/llm.py::get_response_stream | ✅ Since #4735 |
Before #4735 the streaming paths ran tools silently — a streaming UI saw no tool activity even though a tool ran and its result reached the model. The callback now fires once per tool (parallel tools each get their own call), in both the success and failure branches.
tool_call kwargs
A tool_call handler receives these keyword arguments on every path. Accept **kwargs so future additions never break your handler.
| Kwarg | Type | Description |
|---|---|---|
message | str | Human-readable label, e.g. "Calling function: search_web" |
tool_name | str | The function name |
tool_input | dict | Parsed arguments passed to the tool |
tool_output | str or None | json.dumps(result)[:200] (or str(result) if not JSON-serialisable, or None) |
elapsed_time | float or None | perf_counter delta; None on the custom-LLM path when the batch has more than one tool |
success | bool | False for a raising tool and for {"error": ...} results |
Streaming example
tool_call fires the same way whether you stream or not.
from praisonaiagents import Agent, register_display_callback
def on_tool_call(tool_name=None, tool_input=None, tool_output=None,
elapsed_time=None, success=None, **kwargs):
status = "ok" if success else "failed"
secs = f"{elapsed_time:.3f}s" if elapsed_time is not None else "n/a"
print(f"[{status}] {tool_name}({tool_input}) -> {tool_output} in {secs}")
register_display_callback("tool_call", on_tool_call)
def my_tool(query: str) -> str:
"""Look something up."""
return f"Result for {query}"
agent = Agent(
name="researcher",
instructions="Use tools when helpful.",
tools=[my_tool],
)
# Same callback now fires whether stream=True or not.
for chunk in agent.start("Look up the weather in Paris", stream=True):
print(chunk, end="", flush=True)Configuration Options
| Option | Type | Default | Description |
|---|---|---|---|
display_type | str | — | One of the supported callback types above |
callback_fn | callable | — | Sync or async handler |
is_async | bool | False | Store in the async registry when True |
Callbacks receive keyword arguments matching the event (for example message, response, tool_name, agent_name). Extra kwargs are filtered to the function signature automatically.
Best Practices
Keep handlers lightweight
Display callbacks run on the hot path. Log or enqueue work — avoid heavy I/O in sync handlers. The streaming path fires tool_call synchronously inline, so a slow handler blocks token yielding until it returns.
Use async for network and UI
Set is_async=True when writing to WebSockets, Slack, or dashboards so the agent loop stays responsive.
Wrap custom logic in try/except
The agent is already protected — a raising callback is logged and ignored. Add your own try/except only when you want the callback to report its own failures (metrics, alerts) instead of relying on the log line.
Compose with hooks for fine control
Use display callbacks for output formatting; use hooks when you need to allow, block, or mutate tool calls.