axonpush
Python SDKIntegrations

Print capture

Tee sys.stdout and sys.stderr so every print() line from an agent lands on the trace timeline.

pip install axonpush

No extra required, stdlib only.

Plenty of agent code narrates itself with print(). This wraps sys.stdout and sys.stderr in a tee that writes to the real console first, then publishes each completed line as a log event, so that narration shows up in the timeline next to the structured events.

Set it up

from axonpush import AxonPush
from axonpush.integrations.print_capture import setup_print_capture

client = AxonPush()

handle = setup_print_capture(
    client,
    channel_id,
    agent_id="my-agent",
    service_name="my-agent",
)

print("agent starting up")
print("step 1: loaded tools = ['web_search', 'calculator']")

import sys
print("warning: retrying after 429", file=sys.stderr)

client and channel_id are positional; everything else is keyword-only.

The tee writes to the original stream first, so your terminal output is unchanged. Output is buffered until a newline and emitted one event per line, so a multi-write print() does not fragment across events.

Constructor

setup_print_capture(
    client,                  # AxonPush or AsyncAxonPush - positional
    channel_id,              # str UUID - positional
    *,
    agent_id=None,
    source="agent",          # "agent" -> agent.log, "app" -> app.log
    service_name=None,
    mode=None,               # "background" (default) | "sync"
    queue_size=1000,
    shutdown_timeout=2.0,
)

Note the default source is "agent" here, not "app" as in the other logging integrations, this exists for agent output.

Restore the streams

handle.unpatch()

The handle is also a context manager, which is the safer shape:

with setup_print_capture(client, channel_id, agent_id="my-agent"):
    run_agent()
# streams restored

Restore the streams before spawning subprocesses or tearing down a test. An atexit hook unpatches anything still live at interpreter exit, so a crashing process cannot leave a dangling tee, but that is a backstop, not a substitute for scoping it.

unpatch() also closes the background publisher. handle.flush(timeout=...) drains the queue without restoring the streams.

What each line becomes

FieldValue
identifierThe literal print
event_typeagent.log, or app.log when source="app"
payload.bodyThe captured line, without the trailing newline
payload.severityNumber / payload.severityText9 / INFO for stdout, 17 / ERROR for stderr
payload.timeUnixNanoCapture time, in nanoseconds
payload.attributeslog.iostream (stdout or stderr) and log.source (print)
payload.resourceservice.name when configured

Blank lines are skipped. Both streams share one publisher and one trace context, so an agent’s stdout and stderr interleave correctly in the timeline.