Quickstart

Install the published package:

python -m pip install agentlogsafe

Create a logger with a JSONL path and record one workflow run:

from agentlogsafe import AgentLogger

log = AgentLogger(
    agent_name="pricing-agent",
    workflow_name="margin-risk-review",
    sink="agent_events.jsonl",
)

with log.run(input={"request": "Find margin risks in this forecast"}):
    log.user_message("Find margin risks in this forecast")
    log.model_call(
        provider="azure_openai",
        model="gpt-4.1",
        input_tokens=812,
        output_tokens=132,
        payload={"prompt": "Find margin risks in this forecast"},
    )
    log.tool_call(
        tool_name="snowflake.query",
        args={
            "sql": "select * from customer_margin where email = 'person@example.com'",
            "authorization": "Bearer abc.def.ghi",
        },
    )
    log.tool_result(
        tool_name="snowflake.query",
        result={"rows": 20, "sample_email": "person@example.com"},
    )
    log.policy_check(
        policy_name="pii-redaction",
        status="success",
        payload={"contains_pii": True},
    )
    log.decision(
        decision="Escalated because margin impact exceeded threshold",
        requires_human_review=True,
    )

A path automatically creates an append-only JSONLSink. The email addresses and authorization value are redacted before writing:

{"schema_version":"1.0","event_type":"tool_call","tool_name":"snowflake.query","payload":{"args":{"sql":"select * from customer_margin where email = '[REDACTED]'","authorization":"[REDACTED]"}}}

Actual records also include event and run IDs, UTC timestamps, workflow context, status, risk, and metadata.

Trace a Python tool

@log.trace_tool("snowflake.query")
def run_query(sql: str):
    return {"rows": 20}

result = run_query("select * from customer_margin")

The decorator emits tool_call, then tool_result or error, and preserves the function's return value and metadata.

Trace a model call

with log.trace_model_call(
    provider="azure_openai",
    model="gpt-4.1",
    payload={"prompt": "Find margin risks"},
) as trace:
    response = client.responses.create(...)
    trace.set_response({"response_id": response.id})

The context emits model_call immediately. Calling set_response adds a correlated model_response; an exception adds error and is re-raised.

In-memory use

For tests and notebooks, retain events without writing a file:

from agentlogsafe import AgentLogger, MemorySink

sink = MemorySink()
log = AgentLogger(sink=sink)
log.user_message("hello")
assert sink.events[0].event_type == "user_message"

Omitting a sink uses NullSink: events are constructed and returned but discarded.