Anthropic
Wrap messages.create calls to trace the request, the content blocks that come back, and token usage.
pip install "axonpush[anthropic]"Tested against anthropic>=0.30.0,<2.0. The integration reads only documented
public attributes, so any 0.x release with that shape works.
Wrap the call
AxonPushAnthropicTracer does not monkey-patch the Anthropic client. You hand
it the client and it makes the call for you, publishing around it.
import anthropic
from axonpush import AxonPush
from axonpush.integrations.anthropic import AxonPushAnthropicTracer
anthropic_client = anthropic.Anthropic()
client = AxonPush()
tracer = AxonPushAnthropicTracer(client, channel_id, agent_id="claude")
response = tracer.create_message(
anthropic_client,
model="claude-sonnet-4-20250514",
max_tokens=1024,
messages=[{"role": "user", "content": "Summarise this paper."}],
)
tracer.flush(timeout=2.0)create_message and acreate_message take the Anthropic client as the first
argument and forward every remaining keyword to messages.create untouched,
returning its response unchanged. Swapping a call over is a one-line edit:
response = anthropic_client.messages.create(model=..., messages=[...])
response = tracer.create_message(anthropic_client, model=..., messages=[...])One tracer handles both flavours. It inspects the axonpush client you passed
and builds the matching publisher, so use create_message with an AxonPush
and acreate_message with an AsyncAxonPush.
Constructor
AxonPushAnthropicTracer(
client, # AxonPush or AsyncAxonPush - positional
channel_id, # str UUID - positional
*,
agent_id="claude",
trace_id=None,
mode=None, # "background" (default) | "sync"
queue_size=1000, # sync client
shutdown_timeout=2.0, # sync client
max_pending=1000, # async client
)What gets published
| Identifier | Event type | Payload | When |
|---|---|---|---|
conversation.turn | agent.start | model, message_count, max_tokens | Before the request goes out |
agent.usage | agent.message | input_tokens, output_tokens, model, stop_reason | Response carried a usage block |
agent.response | agent.message | text_length | Per text content block |
tool.<name>.start | agent.tool_call.start | tool_name, tool_use_id, input (first 500 chars) | Per tool_use content block |
tool.result | agent.tool_call.end | tool_use_id, result_preview (first 500 chars) | You report the result back |
Every event carries framework: "anthropic" in metadata.
Response text is never published, only its length. Tool inputs and results are truncated to 500 characters.
Report tool results
The Anthropic API has no callback for “the tool ran”, so close the loop yourself when you send the result back into the conversation:
tracer.send_tool_result("toolu_01A...", result)
await tracer.asend_tool_result("toolu_01A...", result) # asyncPass the tool_use_id from the tool_use block so the start and end events
join up.
Flush before you exit
tracer.flush(timeout=2.0) # sync
tracer.close()
await tracer.aflush(timeout=2.0) # async
await tracer.aclose()