agenttrace-langchain

Deprecated. This package is now a thin shim. Use pip install 'lanetrace[langchain]' and import lanetrace.langchain (or lanetrace.langchain.autolog()). Existing top-level imports keep working and emit a DeprecationWarning, as do the old submodule paths (agenttrace_langchain.client, .async_client, .run, .async_run, .middleware, .callback, .projection), which are now aliases.

agenttrace_langchain	New import
AgentTraceMiddleware	lanetrace.langchain.LaneTraceMiddleware
AgentTraceCallbackHandler (0.2.2, async)	lanetrace.langchain.AsyncLaneTraceCallbackHandler
AgentTraceClient	lanetrace.tracking.http.HttpEventClient
AgentTraceRun	lanetrace.tracking.http.LaneTraceRun
AsyncAgentTraceClient	lanetrace.tracking.async_http.AsyncHttpEventClient
AsyncAgentTraceRun	lanetrace.tracking.async_run.AsyncLaneTraceRun

Note: lanetrace.langchain.LaneTraceCallbackHandler (no Async) is the sync handler used by autolog(), not the 0.2.2 class.

0.4.0 depends on lanetrace (AgentTrace / deepagents-trace was renamed LaneTrace in 0.3.0); the agenttrace_langchain names are unchanged. AGENTTRACE_URL / AGENTTRACE_KEY still work as fallbacks for the LANETRACE_* variables below.

Upgrading from 0.2.x

• No API key now posts keylessly. An AsyncAgentTraceClient with no key (api_key=None and no LANETRACE_KEY) used to disable tracing; it now sends without an Authorization header, for a local lanetrace ui --auth none. A server that wants a key answers 401, and the run is then disabled with one warning. If you relied on a missing key as an off switch, gate tracing in your app (don't attach the handler).
• The anonymizer also receives strings. It is called with every payload (a JSON-safe dict), every label (a str) and the run name, and must handle both. If it raises, or returns a non-str for a label, the field is redacted instead of being sent unmasked.
• Only a tool named exactly task (or containing handoff/delegate) counts as a sub-agent handoff; e.g. create_task is a plain tool call.

agenttrace-langchain

AgentTraceMiddleware — an AgentMiddleware for LangChain's create_agent / deepagents' create_deep_agent that streams a run to an LaneTrace instance as a live sequence diagram.

Install

pip install 'lanetrace[langchain]'

This shim package stays installable (pip install -e integrations/agenttrace-langchain) for code that still imports agenttrace_langchain.

Usage

from agenttrace_langchain import AgentTraceMiddleware
from deepagents import create_deep_agent

agent = create_deep_agent(
    model=model,
    tools=[...],
    middleware=[AgentTraceMiddleware(run_name="research run")],
)
agent.invoke({"messages": [{"role": "user", "content": "..."}]})

Works the same with await agent.ainvoke(...) — the middleware's async hooks fire automatically. See examples/ for full sync and async runnable projects, and the offline notebooks in python-server/examples/notebooks/ (written for the new lanetrace.langchain imports).

Configure the target instance and project API key (from the LaneTrace Integration tab, prefixed atr_) via environment variables, or pass them explicitly to the middleware:

middleware = AgentTraceMiddleware(
    run_name="research run",
    url="https://your-deployment/api/events",
    api_key="atr_...",
    timeout=10.0,
)

Reliability

The middleware is built to never affect the agent it instruments:

• Non-blocking — every event is pushed onto a queue drained by a background thread; wrap_model_call/wrap_tool_call never wait on the LaneTrace HTTP call.
• Never fatal — a missing API key or the first network/HTTP failure disables tracing for that run (one warning logged via the standard logging module), the agent keeps running normally.
• Bounded payloads — tool args/results, LLM output previews and the final answer are truncated before being sent, so a single large value can't blow up an event body.

AgentTraceRun implements this contract and can be reused directly if you're not going through the middleware (e.g. to trace a custom orchestration loop):

from agenttrace_langchain import AgentTraceClient, AgentTraceRun

run = AgentTraceRun("custom run", client=AgentTraceClient(api_key="atr_..."))
run.emit(source="Orchestrator", target="LLM", type="llm_call", label="step")
run.end("completed")
run.close()  # bounded wait for the queue to drain

AgentTraceClient itself stays a thin, fail-loud HTTP primitive (raises on a missing key or a failed request) — the right behavior when called directly; AgentTraceRun is the layer that adds the non-blocking/never-fatal guarantees on top.

Servers with a cached/reused agent (don't use the middleware)

AgentTraceMiddleware is baked into the agent graph at create_deep_agent(..., middleware=[...]) build time. If your server compiles the agent once and reuses it across many requests (e.g. a per-user compiled-graph cache with a TTL), a single middleware instance would span multiple runs — the first run's after_agent closes the LaneTrace run, and every later request on that cached agent silently traces into an already-closed run. The middleware is only correct when the agent is (re)built per invocation.

The least-code option is one AgentTraceCallbackHandler (0.2.2 API, now lanetrace.langchain.AsyncLaneTraceCallbackHandler) per request, attached via config={"callbacks": [handler]}; see "Async servers" in the lanetrace README.

Lower-level: create one AsyncAgentTraceRun per request instead, independent of the agent build, and feed it events from your own stream projection (agent.astream_events(...)) rather than from middleware:

from agenttrace_langchain import AsyncAgentTraceClient, AsyncAgentTraceRun

client = AsyncAgentTraceClient(api_key="atr_...")  # reuse across runs; own httpx.AsyncClient optional
run = AsyncAgentTraceRun("chat run", client=client, tool_server=my_mcp_routing_fn)

run.on_user_message(user_text)
async for kind, source, data in my_stream_projection(agent, ...):
    run.on_stream_event(kind, source, data)  # tool_call/tool_result/agent_start/agent_end/approval_required/final
run.end("completed")
await run.aclose()

tool_server is an optional Callable[[str], str] — pass it if you want a payload.server label on tool arrows (e.g. which MCP/backend served a tool call); omit it if you don't need that. Note on_stream_event's diagram labels ("delegate → X", "failed"/"done") are in English and not currently customizable — fork or post-process if you need different wording.

Token usage still needs a BaseCallbackHandler (attach via config={"callbacks": [...]} at invoke time) since a stream projection typically doesn't expose LLM call boundaries — callbacks, unlike middleware, correctly compose with a cached/reused agent because they're attached per-invocation rather than baked into the graph.

Event mapping

Tests

pip install -e ".[dev]"
pytest