Skip to main content

Telemetry

The Telemetry subsystem provides standard observable tracing to agent task execution, tool calling, and LLM inference. Protolink supports a non-invasive integration with external tracing services using Python's contextvars. This means it tracks nested traces, spans, and runs in the background without cluttering core execution method signatures.

Protolink includes a built-in local trace recorder and native integrations for Langfuse and LangSmith.

Observability moduleTelemetry

The tracing layer for task runs, tool calls, LLM spans, context usage, cost estimates, redacted local traces, and optional Langfuse or LangSmith export.

protolink.telemetry
LocalTraceTelemetryLangfuseLangSmithLLM metricsRedactionPolicy
Trace locallyRecord nested spans and replayable JSONL traces without requiring an external service.LocalTraceTelemetry
ExportSend compatible trace data to Langfuse or LangSmith when those optional integrations are installed.protolink[telemetry]
Measure modelsCapture context pressure, latency, token usage, and estimated cost around LLM calls.llm_call_metrics
Protect dataApply the shared redaction policy before common secrets are persisted or exported.RedactionPolicy

How telemetry fits into execution​

Telemetry is an Agent lifecycle boundary, not a replacement for runtime events or reports. When agent.telemetry is set, the Agent awaits coarse hooks around the task, each direct task-level tool call, and the complete LLM.infer() cycle. While inference is running, it also forwards provider-neutral loop events to on_llm_event().

on_task_start
├── on_tool_start → on_tool_end
└── on_llm_start
    ├── on_llm_event: context_prepared
    ├── on_llm_event: llm_call_metrics
    ├── on_llm_event: tool_start → tool_result | tool_error
    ├── on_llm_event: agent_call_start → agent_call_result | agent_call_error
    └── on_llm_end
on_task_end

The exact middle events depend on the action loop. A simple inference can emit context, call, response, action, and final events; multi-step inference can add streamed chunks, retries, tool operations, delegated-agent operations, budget decisions, and more. LocalTraceTelemetry retains those detailed events. The hosted backends currently inherit the base no-op implementation of on_llm_event(), so Langfuse and LangSmith receive the coarse task, LLM, and explicit tool lifecycle only.

Telemetry is non-authoritative when attached to an Agent. The runtime catches hook exceptions, logs the first failure for each hook name, and continues with the task, LLM, or tool result unchanged. This isolation applies to unary and streaming task lifecycles. Calling a telemetry implementation's hook directly still follows that implementation's own exception contract.

Telemetry versus runtime reporting

Use telemetry for detailed traces, span hierarchy, observability export, and local debugging. Use RunEvent, RunRecorder, and RunReport from the Runtime layer for the stable application-facing event envelope, durable run summaries, replay, and regression assertions. An application can use both surfaces on the same Agent.

Direct LLM calls

Attaching telemetry to an Agent instruments calls made through that Agent. A direct llm.infer() call does not discover an Agent's telemetry object; pass an event_callback when you need its live provider-neutral events outside Agent execution.

Installation​

Telemetry dependencies are handled as optional plugins. To use a telemetry integration, you must install its corresponding library:

# Install telemetry with langfuse and langsmith
uv add "protolink[telemetry]"

# Or install telemetry with just langfuse
uv add langfuse
# Or install telemetry with just langsmith
uv add langsmith

# Optional: improve local token estimates for LLM metrics
uv add "protolink[metrics]"

Setup & Usage​

To enable observability, instantiate your preferred telemetry tracker and inject it into your Agent. Tasks executed by this agent will now automatically trace their internal states and synchronize with your observability platform.

Local Trace Example​

LocalTraceTelemetry records task traces in memory and can append replayable JSONL records to disk. It captures trace IDs, parent-child spans, model metadata, token estimates, raw inference-loop events, retry counts, and redacted payloads without requiring an external service.

from protolink import Agent, AgentCard, LocalTraceTelemetry, Task

telemetry = LocalTraceTelemetry(path="traces.jsonl")

agent = Agent(
    card=AgentCard(
        name="local_observer",
        description="A locally traced agent",
        url="runtime://local-observer",
    ),
    telemetry=telemetry,
)

@agent.tool(name="add", description="Add two integers")
async def add(a: int, b: int) -> int:
    return a + b

result = await agent.handle_task(Task.create_tool_call(tool_name="add", args={"a": 2, "b": 3}))
records = telemetry.recorder.replay()
View traces in Devtools

Open the persisted telemetry file as a timeline and span waterfall:

protolink dashboard --traces traces.jsonl --open

--telemetry is an alias for --traces, and the dashboard Telemetry view also accepts a JSONL file selected locally in the browser. It pages recent task records, rolls a bounded summary window through older history, and loads detail payloads lazily rather than reading the entire file into the initial page. See Developer Tools for shared trace_id grouping, scan and detail safeguards, partial-line handling, and local-data security guidance.

LLM Metrics and Context Usage​

When an agent has both an LLM and telemetry, Protolink records live context and budget metadata for every model call inside LLM.infer(). This includes the pre-call context manifest, latency, token usage, context-window pressure, and estimated cost. Provider-reported usage is used when available; otherwise Protolink estimates usage without requiring extra dependencies.

from protolink import Agent, AgentCard, LLMModelProfile, LocalTraceTelemetry, Task, create_llm

telemetry = LocalTraceTelemetry(path="traces.jsonl")
llm = create_llm(
    "openai-compatible",
    model="my-model",
    metrics_profile=LLMModelProfile(
        context_window=128_000,
        input_cost_per_million=1.0,   # example value; use your provider's current pricing
        output_cost_per_million=5.0,  # example value; use your provider's current pricing
    ),
)

agent = Agent(
    card=AgentCard(name="budgeted", description="Observed LLM agent", url="runtime://budgeted"),
    llm=llm,
    telemetry=telemetry,
)

result = await agent.handle_task(Task.create_infer(prompt="Plan the next release"))
trace = telemetry.recorder.replay()[-1]
llm_span = next(span for span in trace["spans"] if span["kind"] == "llm")
print(llm_span["metadata"]["llm_metrics"])

The same data is emitted live as context_prepared, llm_context, and llm_call_metrics events through event_callback, so terminal apps can render a status line such as context used, call latency, and session cost while the agent is still running.

Local trace telemetry and runtime reports share the same default RedactionPolicy, so common secret-bearing fields such as API keys, tokens, passwords, authorization headers, and credentials are masked consistently before data is persisted.

Cost estimates

Protolink does not ship a fixed provider pricing catalog. Prices and context windows are application-owned metadata passed through LLMModelProfile, which keeps the core package stable and avoids stale billing assumptions.

Langfuse Example​

The LangfuseTelemetry tracks tasks as traces, and LLM/Tool executions as spans/generations.

import os
from protolink.telemetry import LangfuseTelemetry
from protolink.agents.base import Agent

# Ensure environment variables are set:
# os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..."
# os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..."
# os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"

# Initialize tracking
telemetry_tracker = LangfuseTelemetry()

# Inject into an agent
agent = Agent(
    card={"name": "ObserverAgent", "description": "Observed agent", "url": "runtime://observer"},
    telemetry=telemetry_tracker
)

LangSmith Example​

The LangSmithTelemetry uses the RunTree API to track tasks hierarchically.

import os
from protolink.telemetry import LangSmithTelemetry
from protolink.agents.base import Agent

# Ensure environment variables are set:
# os.environ["LANGCHAIN_API_KEY"] = "lsv2_pt_..."
# os.environ["LANGCHAIN_PROJECT"] = "my-protolink-project"

# Initialize tracking
telemetry_tracker = LangSmithTelemetry()

# Inject into an agent
agent = Agent(
    card={"name": "ObserverAgent", "description": "Observed agent", "url": "runtime://observer"},
    telemetry=telemetry_tracker
)

Multiplexing Telemetry​

If you want to broadcast telemetry events to multiple trackers simultaneously, you can use the MultiTelemetry class:

from protolink.telemetry import LangfuseTelemetry, LangSmithTelemetry, MultiTelemetry
from protolink.agents.base import Agent

# Initialize tracking
langfuse_tracker = LangfuseTelemetry()
langsmith_tracker = LangSmithTelemetry()
multi_tracker = MultiTelemetry([langfuse_tracker, langsmith_tracker])

# Inject into an agent
agent = Agent(
    card={"name": "ObserverAgent", "description": "Observed agent", "url": "runtime://observer"},
    telemetry=multi_tracker
)

MultiTelemetry awaits trackers sequentially in list order. It does not isolate failures or run trackers concurrently: if a custom tracker raises, later trackers do not receive that hook and the exception reaches the Agent. The built-in Langfuse and LangSmith lifecycle methods catch provider-operation failures and log warnings, but constructor and dependency errors still propagate.

Setting Telemetry Dynamically​

You can also change or assign a telemetry tracker after agent initialization using the .telemetry property:

agent = Agent(card={"name": "ObserverAgent", "description": "Observed agent", "url": "runtime://observer"})

# Later in your code...
agent.telemetry = LangfuseTelemetry()

The setter accepts a Telemetry instance or None and performs no runtime validation. Change it between tasks whenever possible. Replacing a tracker during an active task can give the new tracker an end hook without its matching start hook, while the previous tracker keeps unfinished provider context.

Creating Custom Telemetry Implementations​

If you wish to integrate with a different observability platform (e.g., Datadog, Prometheus, Arize Phoenix), subclass Telemetry and implement its six abstract asynchronous hooks. Override on_llm_event() only when the backend also needs detailed inference-loop events; the base implementation deliberately returns None.

from typing import Any
from protolink.models import Task, Part
from protolink.telemetry.base import Telemetry

class MyCustomTelemetry(Telemetry):
    async def on_task_start(self, task: Task, agent_name: str) -> Any:
        pass
        
    async def on_task_end(self, task: Task, result: Task, agent_name: str) -> Any:
        pass
        
    async def on_llm_start(
        self,
        prompt: str,
        model: str | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> Any:
        pass
        
    async def on_llm_end(self, response: Part) -> Any:
        pass
        
    async def on_tool_start(self, tool_name: str, args: dict[str, Any]) -> Any:
        pass
        
    async def on_tool_end(self, tool_name: str, result: Any, error: str | None = None) -> Any:
        pass

    async def on_llm_event(self, event: dict[str, Any]) -> Any:
        pass

Hooks are awaited inline by the Agent. A custom implementation should therefore avoid blocking I/O in the event loop, maintain per-task state with contextvars or another concurrency-safe mechanism, and decide explicitly whether export failures should propagate or be converted into warnings. Return values are permitted by the abstract contract but are ignored by the current Agent runtime.

Example Code​

Here is a complete example demonstrating telemetry tracking with an Agent using Langfuse:

import asyncio

from protolink import Agent, AgentCard, Task, create_llm
from protolink.telemetry import LangfuseTelemetry


async def main() -> None:
    # OPENAI_API_KEY and the Langfuse environment variables must be set.
    llm = create_llm("openai", model="gpt-4o-mini")
    telemetry = LangfuseTelemetry()

    agent = Agent(
        card=AgentCard(
            name="HelperAgent",
            description="Observed helper",
            url="runtime://helper",
        ),
        llm=llm,
        telemetry=telemetry,
    )

    result = await agent.handle_task(
        Task.create_infer(prompt="Give me one concise release-planning tip.")
    )
    print(result.get_last_part_content())


asyncio.run(main())

Telemetry API reference​

The package-level public surface is:

from protolink.telemetry import (
    LangfuseTelemetry,
    LangSmithTelemetry,
    LocalTraceRecorder,
    LocalTraceTelemetry,
    MultiTelemetry,
    Telemetry,
    TraceEvent,
    TraceRecord,
    TraceSpan,
)

LocalTraceRecorder and LocalTraceTelemetry are also available from the top-level protolink package. The hosted providers, multiplexer, abstract contract, and trace dataclasses are imported from protolink.telemetry.

Telemetry​

abstract classprotolink.telemetry.Telemetry
source
class Telemetry()

Define the asynchronous lifecycle contract shared by every telemetry backend. The class stores no state and supplies no constructor arguments. Concrete implementations decide how to preserve task-local hierarchy, serialize values, export data, and handle backend failures.

Abstract methods

on_task_startasync methodrequired

Begin task-level telemetry before the Agent executes the task.

on_task_endasync methodrequired

Finalize task-level telemetry with the Task returned by execution, or with the original Task on the Agent's exception path.

on_llm_startasync methodrequired

Begin the span or run that surrounds one complete LLM.infer() operation.

on_llm_endasync methodrequired

Finalize that LLM operation after inference returns a response Part.

on_tool_startasync methodrequired

Begin an explicit task-level tool operation.

on_tool_endasync methodrequired

Finalize an explicit task-level tool operation with either a result or an error string.

Concrete extension hook

on_llm_eventasync method

Optional detailed inference-event hook. Its base implementation is a no-op returning None, so older or coarse-grained providers remain instantiable without implementing it.

Raises

TypeError

Instantiating Telemetry directly, or instantiating a subclass that has not implemented all six abstract methods, fails through Python's ABC machinery.

Inline execution

Agent awaits every hook in its execution path; hook return values are currently ignored. The abstract return type is Any so providers may return context for direct callers, but implementations that perform slow synchronous export work can delay the task.

Local tracing API​

The local backend is dependency-free. One LocalTraceTelemetry instance manages lifecycle state; its LocalTraceRecorder owns completed records in memory and, optionally, appends them to JSONL. TraceRecord, TraceSpan, and TraceEvent are the structured objects retained in memory.

default_redactor​

functionprotolink.telemetry.local.default_redactor
source
default_redactor(
  value: Any,
) -> Any

Best-effort normalize common runtime values and recursively mask fields recognized by the shared DEFAULT_REDACTION_POLICY.

Parameters

valueAnyrequired

Nested runtime value. Objects with to_dict(), dataclass instances, mappings, lists, tuples, and sets receive special handling; ordinary non-JSON-native leaves fall back to str(value).

Returns

redactedAny

Best-effort normalized value with case-insensitive secret fields masked. Default sensitive names include API key, authorization, client secret, credentials, password, secret, and token variants, including common suffixed forms.

Raises

serialization or user-object error

Exceptions raised by an object's custom to_dict() or dataclass conversion are not caught.

Import path

This helper is public in protolink.telemetry.local but is not re-exported by protolink.telemetry. Most applications configure redaction through LocalTraceTelemetry(redactor=...) instead of calling it directly.

Custom serializer boundary

A custom to_dict() result or a dataclass field is not passed through a second complete normalization cycle. If it contains an unsupported leaf object, later JSONL encoding can still raise TypeError.

LocalTraceTelemetry​

classprotolink.telemetry.LocalTraceTelemetry
source
LocalTraceTelemetry(
  recorder: LocalTraceRecorder | None = None,
  *,
  path: str | Path | None = None,
  redactor: Callable[[Any], Any] | None = None,
  capture_payloads: bool = True,
  max_traces: int = 1000,
) -> None

Record one replayable local trace per completed Agent task. Task, LLM, explicit tool, and inference-selected child operations are connected through IDs stored in contextvars, so concurrent async task contexts do not need trace objects passed through every runtime call.

Parameters

recorderLocalTraceRecorder | Nonedefault: None

Existing recorder to use. Supplying one is useful for sharing completed trace storage across agents or tests. Because the implementation selects it with recorder or LocalTraceRecorder(...), a custom recorder with false truthiness is replaced rather than retained.

pathstr | Path | Nonedefault: None

JSONL destination used only when the constructor creates its own recorder. It is ignored when a truthy recorder is supplied. Parent directories are created on the first completed trace.

redactorCallable[[Any], Any] | Nonedefault: None

Application-specific transformation applied after default normalization and secret masking at explicit capture points: span inputs and outputs, event payloads, span metadata, and context or budget mappings. Generated trace status and metric-rollup fields are assigned directly.

capture_payloadsbooldefault: True

When true, retain span inputs and outputs plus event payloads. When false, those values become None or an empty mapping; span and trace metadata, event types, timing, IDs, statuses, and metrics are still recorded.

max_tracesintdefault: 1000

In-memory retention limit passed to the automatically created recorder. Positive values keep only the newest records. Zero and negative values disable truncation, not recording.

Attributes

recorderLocalTraceRecorder

Recorder receiving a trace when on_task_end() completes.

redactorCallable[[Any], Any] | None

Exact custom callable supplied at construction.

capture_payloadsbool

Current payload-capture switch. It is a normal mutable attribute, so applications can change it, although changing it during a task can produce a mixed-detail trace.

Raises

constructor error

The constructor performs no explicit validation. Most path, redactor, serialization, and persistence errors occur later in lifecycle hooks and propagate to the Agent.

Redaction order

Default redaction always runs first. A custom redactor can remove more data, but it receives already masked values and cannot recover secrets. Its return value is not normalized again, so returning a non-JSON-serializable object can make JSONL persistence fail.

Examples

from protolink import LocalTraceRecorder, LocalTraceTelemetry

shared = LocalTraceRecorder(path="var/traces.jsonl", max_traces=250)
telemetry = LocalTraceTelemetry(
    recorder=shared,
    capture_payloads=True,
    redactor=lambda value: value,
)

LocalTraceTelemetry lifecycle methods​

async methodsprotolink.telemetry.LocalTraceTelemetry lifecycle
source
async on_task_start(task: Task, agent_name: str) -> Any
async on_task_end(task: Task, result: Task, agent_name: str) -> Any
async on_llm_start(
  prompt: str,
  model: str | None = None,
  metadata: dict[str, Any] | None = None,
) -> Any
async on_llm_end(response: Part) -> Any
async on_tool_start(tool_name: str, args: dict[str, Any]) -> Any
async on_tool_end(
  tool_name: str,
  result: Any,
  error: str | None = None,
) -> Any
async on_llm_event(event: dict[str, Any]) -> Any

Implement the complete telemetry contract and translate it into a local trace hierarchy. All seven methods return None implicitly; lifecycle state is held in the current context and the finished TraceRecord is committed through the recorder.

Task lifecycle

on_task_start

Reuse a truthy task.metadata["trace_id"] or generate a UUID, write it back to the Task, create a running TraceRecord, reset the context-local span stack, and open a root kind="task" span containing the serialized task.

on_task_end

If no trace is active, return without action. Otherwise close the nearest active task span, derive error state only from result.metadata["error"], set final-state and retry metadata, append the record through the recorder, and clear local context.

LLM lifecycle

on_llm_start

Open one kind="llm" span named "LLM Call". Metadata includes the model, any supplied cost field, prompt character count, a four-character token estimate, and the caller metadata merged afterward.

on_llm_end

Close the nearest active LLM span, capture response content, and add output character count plus the same local token estimate.

Explicit tool lifecycle

on_tool_start

Open a kind="tool" child span with the requested arguments and metadata identifying source="task".

on_tool_end

Close the nearest active tool span. A truthy error marks it status="error" and stores the message; otherwise status remains "ok", even when the result is None.

Detailed inference events

context_prepared

Store a mapping-valued manifest in the current span's context_manifest metadata, then append the chronological event.

llm_context

Store a mapping-valued context in the current span metadata, then append the event.

budget_warning | budget_exceeded

Append a mapping-valued decision to trace-level budget_decisions, then append the event.

llm_call_metrics

Aggregate call count, latency, usage, estimated-call count, context pressure, window size, cost, and currency into both the current span and the trace before appending the event.

tool_start

Open a nested kind="tool" span with source="llm_loop", tool name, step, and arguments.

tool_result | tool_error

Append the event once, close the nearest tool span, attach result/name/step, and mark an error from message for tool_error.

agent_call_start

Open a nested kind="agent_call" span containing agent, action, step, and payload.

agent_call_result | agent_call_error

Append the event once and close the nearest delegated-agent span, using message as its error when appropriate.

llm_parse_error

Set trace retry count to the maximum of its existing value and the event's retry_count, falling back to parse_failures.

llm_retry

Increment trace retry count by one.

other event types

Preserve the full redacted event chronologically. A missing type is normalized to "llm_event".

Fallback behavior

no active trace

LLM, tool, and event hooks return without recording anything.

no matching active span

Span-closing calls return without error. Events still attach to the trace and to the current span when one exists.

token usage unavailable

Prompt and response estimates use max(1, len(str(value)) // 4) for non-empty content and zero for empty content. This tracer-level estimate does not use tiktoken; provider-normalized metrics can separately carry richer usage.

non-numeric metrics

Values that cannot be converted with float() are skipped during aggregation rather than raising.

Raises

redactor error

Exceptions from the custom redactor propagate when the hook is called directly. Agent catches and logs them at its observability boundary.

serialization or filesystem error

Task/result conversion, malformed metric values used by explicit integer conversion, directory creation, JSON encoding, and file writes may propagate from a direct hook call. Task-end cleanup runs in finally, so the active local trace frame is still cleared when recording fails; Agent execution also isolates that failure.

Unfinished child spans

If inference raises, Agent does not call on_llm_end(). The subsequent task-end hook closes the root task span by kind but does not synthesize an end time or error for the still-open LLM child span, so replay can contain a child with ended_at=None.

Base lifecycle method reference​

Telemetry.on_task_start​

abstract async methodprotolink.telemetry.Telemetry.on_task_start
source
async on_task_start(
  task: Task,
  agent_name: str,
) -> Any

Receive control immediately before an Agent begins executing a task. Implementations normally allocate a root trace or run and bind it to the current async context.

Parameters

taskTaskrequired

The live mutable Task about to execute. A backend may inspect its ID, state, parts, metadata, and attached run context. Mutating it affects the task seen by the runtime; LocalTraceTelemetry intentionally adds or reuses task.metadata["trace_id"].

agent_namestrrequired

The current Agent card's name. It is not independently normalized or validated by the hook contract.

Returns

contextAny

Optional backend-specific context. The current Agent discards this value, so built-in implementations return None implicitly and retain state with contextvars.

Raises

implementation error

A direct call can propagate an implementation error. Agent invokes the hook through its best-effort telemetry boundary, logs the first failure for this hook name, and continues task execution.

Telemetry.on_task_end​

abstract async methodprotolink.telemetry.Telemetry.on_task_end
source
async on_task_end(
  task: Task,
  result: Task,
  agent_name: str,
) -> Any

Finalize the active task trace after execution. On the normal path, result is the Task returned by execute_task(). When execution raises, Agent invokes the hook with the original task as both task and result, then re-raises the execution exception.

Parameters

taskTaskrequired

Original task object passed to Agent.handle_task().

resultTaskrequired

Completed Task on success. On the current exception path this is the original Task, which may not contain an error metadata field.

agent_namestrrequired

Name of the Agent that handled the task.

Returns

contextAny

Optional provider value; ignored by Agent. Built-in backends return None implicitly.

Failure-path detail

A backend cannot infer every execution exception from result.metadata. In particular, the local backend marks a trace as failed only when that metadata contains a truthy error value. A raised execution error without that metadata can therefore produce a locally recorded trace whose status is "ok".

Hook failure isolation

Agent calls this hook once from its task-finalization boundary. A hook failure is logged and does not replace the task result or the execution exception already in flight.

Telemetry.on_llm_start​

abstract async methodprotolink.telemetry.Telemetry.on_llm_start
source
async on_llm_start(
  prompt: str,
  model: str | None = None,
  metadata: dict[str, Any] | None = None,
) -> Any

Begin observability around one complete LLM.infer() cycle. This is broader than one provider request: a single inference can call the model repeatedly while resolving tools, delegated agents, parse retries, and the final answer.

Parameters

promptstrrequired

User inference prompt extracted from the Task's infer Part. It is the query supplied to LLM.infer(), not the fully compiled provider conversation or system prompt.

modelstr | Nonedefault: None

Best available model identifier, selected from the LLM's model_name or model attribute. It is None when neither exists.

metadatadict[str, Any] | Nonedefault: None

Optional provider context. Agent currently passes agent_name, task_id, trace_id, provider, and model_type. Direct callers may pass a different mapping.

Returns

contextAny

Optional provider generation/span context; ignored by Agent.

No matching end on inference failure

Agent calls on_llm_end() only after LLM.infer() returns. If inference raises, the task-end hook still runs, but an implementation must decide whether and how to close an unfinished LLM span.

Start-hook failure

If this hook raises through Agent, the runtime logs the failure and still starts inference. The failure does not suppress the matching end-hook attempt after a successful infer result.

Telemetry.on_llm_end​

abstract async methodprotolink.telemetry.Telemetry.on_llm_end
source
async on_llm_end(
  response: Part,
) -> Any

Finish the active LLM operation after the controlled inference loop has produced its final Part.

Parameters

responsePartrequired

Final Part returned by LLM.infer(). Hosted providers extract response.content; the local backend stores and estimates tokens from the same content.

Returns

contextAny

Optional backend value; ignored by Agent. Built-ins return None implicitly.

Telemetry.on_tool_start​

abstract async methodprotolink.telemetry.Telemetry.on_tool_start
source
async on_tool_start(
  tool_name: str,
  args: dict[str, Any],
) -> Any

Begin a span for a tool call executed from an explicit task tool_call Part. Tools chosen inside the LLM action loop are surfaced instead through on_llm_event() as tool_start events.

Parameters

tool_namestrrequired

Registered tool name resolved by Agent.

argsdict[str, Any]required

Requested keyword arguments before runtime policy authorization. A policy or approval handler can subsequently modify the actual arguments passed to the tool.

Returns

contextAny

Optional provider span context; ignored by Agent.

Missing tools

Agent resolves the tool before invoking this hook. An unknown tool returns an error Part without producing on_tool_start() or on_tool_end().

Start-hook errors

Agent isolates this observer failure and continues through policy authorization and tool execution. Direct hook calls can still propagate according to the implementation.

Telemetry.on_tool_end​

abstract async methodprotolink.telemetry.Telemetry.on_tool_end
source
async on_tool_end(
  tool_name: str,
  result: Any,
  error: str | None = None,
) -> Any

Finish an explicit task-level tool span. Agent sends the returned value on success; policy failures, cancellation, and tool exceptions are represented by result=None plus a string error.

Parameters

tool_namestrrequired

Registered tool name associated with the active operation.

resultAnyrequired

Tool return value on success. Agent supplies None on its caught error paths, so a successful tool that legitimately returns None is distinguished by the separate error argument.

errorstr | Nonedefault: None

String form of the failure, or None for success.

Returns

contextAny

Optional backend value; ignored by Agent.

End-hook errors

Agent logs and isolates this observer failure. A successful tool result remains successful, a tool error keeps its original error, and the hook is not called a second time merely because telemetry export failed.

Telemetry.on_llm_event​

async methodprotolink.telemetry.Telemetry.on_llm_event
source
async on_llm_event(
  event: dict[str, Any],
) -> Any

Receive a provider-neutral event emitted while LLM.infer() is running. This optional high-detail hook carries context manifests, model-call metrics, chunks, actions, retries, tools, delegated agents, budget decisions, and final outputs without expanding the coarse lifecycle signature.

Parameters

eventdict[str, Any]required

Event mapping whose type key selects its schema. Consumers should tolerate unknown event types and additive fields because providers and inference paths emit different detail.

Returns

NoneNone

The base implementation returns None. Agent ignores return values from overrides.

Hosted-provider behavior

LangfuseTelemetry and LangSmithTelemetry do not override this method in the current implementation. Use MultiTelemetry with a local or custom detailed tracker when you need hosted coarse traces and complete local inference events together.

Observer failure

Agent isolates telemetry-hook exceptions. For direct LLM.infer(event_callback=...) usage, the inference loop logs the first callback exception and disables that callback for the rest of the infer call.

Local recorder API​

LocalTraceRecorder​

classprotolink.telemetry.LocalTraceRecorder
source
LocalTraceRecorder(
  path: str | Path | None = None,
  *,
  max_traces: int = 1000,
) -> None

Retain completed TraceRecord objects in process and optionally append one serialized record per line to a JSONL file.

Parameters

pathstr | Path | Nonedefault: None

Optional JSONL file. Truthy values are converted to Path(path).expanduser(); None and other falsey values such as an empty string disable file persistence.

max_tracesintdefault: 1000

Positive in-memory retention limit. After each append, only the newest max_traces objects remain. Zero or a negative value keeps every trace.

Attributes

pathPath | None

Expanded destination or None.

max_tracesint

Mutable retention value consulted on each subsequent record().

traceslist[TraceRecord]

Live retained record objects in completion order. This list is public and mutable; use replay() when callers need serialized dictionaries.

Memory and disk retention are independent

The retention limit affects only traces. JSONL is append-only: truncating memory or calling clear() never removes lines already written to disk.

LocalTraceRecorder.record​

methodprotolink.telemetry.LocalTraceRecorder.record
source
record(
  trace: TraceRecord,
) -> None

Append a completed trace to memory, enforce the positive retention limit, and then append its dictionary representation to the configured JSONL destination.

Parameters

traceTraceRecordrequired

Record object retained by identity in memory. The recorder performs no runtime type check; file persistence later expects a callable to_dict().

Returns

NoneNone

The method mutates recorder state and has no value return.

Raises

AttributeError

File persistence is enabled and the supplied object has no suitable to_dict().

TypeError

The serialized mapping still contains a value rejected by json.dumps().

OSError

Parent-directory creation or append-mode UTF-8 writing fails.

Mutation before persistence

The trace is appended to memory before the file is opened. If JSON serialization or writing fails, the exception propagates but the in-memory trace remains recorded.

LocalTraceRecorder.replay​

methodprotolink.telemetry.LocalTraceRecorder.replay
source
replay(
  trace_id: str | None = None,
) -> list[dict[str, Any]]

Serialize retained in-memory traces for inspection, tests, or a replay UI. This method does not read the JSONL file.

Parameters

trace_idstr | Nonedefault: None

Exact trace-ID filter. Omit it to return every retained record in completion order. An unknown ID produces an empty list.

Returns

recordslist[dict[str, Any]]

Fresh dictionary serialization for each matching TraceRecord, including computed durations and nested serialized spans/events.

Raises

timestamp or serialization error

Malformed timestamps or manually inserted objects that violate the trace dataclass expectations can fail during to_dict().

LocalTraceRecorder.clear​

methodprotolink.telemetry.LocalTraceRecorder.clear
source
clear() -> None

Remove every in-memory TraceRecord from the recorder.

Returns

NoneNone

The existing list is cleared in place.

JSONL is preserved

This method deliberately does not truncate, replace, or delete the configured file. Use filesystem retention under explicit application control.

LocalTraceRecorder.load_jsonl​

class methodprotolink.telemetry.LocalTraceRecorder.load_jsonl
source
load_jsonl(
  path: str | Path,
) -> list[dict[str, Any]]

Read an existing JSONL trace file without constructing a recorder or mutating in-memory state.

Parameters

pathstr | Pathrequired

File path converted with Path(path).expanduser().

Returns

recordslist[dict[str, Any]]

Decoded non-empty lines in file order. A missing path returns an empty list. Blank or whitespace-only lines are skipped.

Raises

json.JSONDecodeError

Any non-empty malformed JSON line aborts the entire load; partial results are not returned.

OSError | UnicodeError

The path cannot be opened/read as UTF-8.

No schema reconstruction

Each line is returned exactly as json.loads() decodes it. The method does not validate trace keys and does not reconstruct TraceRecord, TraceSpan, or TraceEvent instances.

Examples

from protolink import LocalTraceRecorder

records = LocalTraceRecorder.load_jsonl("~/protolink/traces.jsonl")
failed = [record for record in records if record.get("status") == "error"]

Local trace data model​

TraceEvent​

dataclassprotolink.telemetry.TraceEvent
source
TraceEvent(
  type: str,
  timestamp: str = field(default_factory=_utc_now),
  span_id: str | None = None,
  payload: dict[str, Any] = field(default_factory=dict),
) -> None

Represent one point-in-time inference event. Events live in the trace-wide chronological list and, when a span is active, in that span's event list as well.

Fields

typestrrequired

Event discriminator such as "context_prepared", "llm_call_metrics", "tool_result", or an application-defined value.

timestampstrdefault: current UTC ISO-8601 time

Serialized timestamp generated with timezone-aware datetime.now(timezone.utc).isoformat() when omitted.

span_idstr | Nonedefault: None

ID of the active span when the event was recorded, or None when it belongs only to the trace.

payloaddict[str, Any]default: {}

Event data created with a per-instance default factory. Local telemetry redacts it before construction; direct construction performs no validation or redaction.

Runtime validation

Dataclass annotations are not enforced at runtime. Direct callers can construct inconsistent values; downstream serialization and replay code assumes the documented shapes.

TraceEvent.to_dict​

methodprotolink.telemetry.TraceEvent.to_dict
source
to_dict() -> dict[str, Any]

Convert the event with dataclasses.asdict().

Returns

eventdict[str, Any]

Deep dataclass conversion containing type, timestamp, span_id, and payload.

JSON compatibility

Values captured through LocalTraceTelemetry are normalized before reaching the event. A manually constructed payload is only converted by asdict(); it is not independently redacted or guaranteed JSON-serializable.

TraceSpan​

dataclassprotolink.telemetry.TraceSpan
source
TraceSpan(
  id: str,
  trace_id: str,
  name: str,
  kind: str,
  parent_id: str | None = None,
  started_at: str = field(default_factory=_utc_now),
  ended_at: str | None = None,
  status: str = "ok",
  input: Any | None = None,
  output: Any | None = None,
  error: str | None = None,
  metadata: dict[str, Any] = field(default_factory=dict),
  events: list[TraceEvent] = field(default_factory=list),
) -> None

Represent one timed operation inside a local trace. The built-in tracer uses kind values task, llm, tool, and agent_call; parent IDs encode hierarchy while the record stores spans in a flat list.

Identity and hierarchy

idstrrequired

Unique span ID. Local telemetry generates a UUID.

trace_idstrrequired

Owning trace ID.

namestrrequired

Human-readable operation name such as "LLM Call" or "Tool: add".

kindstrrequired

Machine-readable operation category. Direct construction is not restricted to built-in values.

parent_idstr | Nonedefault: None

Parent span ID. The root task span has no parent.

Timing and outcome

started_atstrdefault: current UTC ISO-8601 time

Start timestamp.

ended_atstr | Nonedefault: None

End timestamp. It remains None for an open or abandoned span.

statusstrdefault: "ok"

Outcome label. Local telemetry changes it to "error" only when ending the span with a truthy error string.

errorstr | Nonedefault: None

Captured error message.

Captured data

inputAny | Nonedefault: None

Redacted operation input when payload capture is enabled.

outputAny | Nonedefault: None

Redacted operation output once closed and when payload capture is enabled.

metadatadict[str, Any]default: {}

Redacted identifiers, source, steps, context manifests, and metric rollups. The default is per instance.

eventslist[TraceEvent]default: []

Detailed events observed while this span was active. The default is per instance.

TraceSpan.duration_ms and TraceSpan.to_dict​

property and methodprotolink.telemetry.TraceSpan inspection
source
duration_ms: float | None
to_dict() -> dict[str, Any]

Inspect elapsed time and serialize the complete span.

Returns

duration_msfloat | None

None while ended_at is absent. Otherwise parse both ISO timestamps, subtract them, convert to milliseconds, and round to three decimal places.

to_dict()dict[str, Any]

All dataclass fields plus computed duration_ms, with each child event serialized through TraceEvent.to_dict().

Raises

ValueError

A non-empty timestamp is not accepted by datetime.fromisoformat().

AttributeError

A manually inserted item in events does not provide to_dict().

No monotonicity validation

The dataclass does not verify that ended_at is later than started_at; directly supplied timestamps can therefore produce a negative duration.

TraceRecord​

dataclassprotolink.telemetry.TraceRecord
source
TraceRecord(
  trace_id: str,
  task_id: str,
  agent_name: str,
  started_at: str = field(default_factory=_utc_now),
  ended_at: str | None = None,
  status: str = "running",
  metadata: dict[str, Any] = field(default_factory=dict),
  spans: list[TraceSpan] = field(default_factory=list),
  events: list[TraceEvent] = field(default_factory=list),
) -> None

Represent the top-level replay artifact for one task. The local backend retains a flat span list with parent IDs and a trace-wide chronological event list.

Identity

trace_idstrrequired

Trace correlation ID, normally reused from or written into task metadata.

task_idstrrequired

Source Task ID.

agent_namestrrequired

Agent name supplied to the task-start hook.

Timing and status

started_atstrdefault: current UTC ISO-8601 time

Trace creation time.

ended_atstr | Nonedefault: None

Completion time assigned by LocalTraceTelemetry.on_task_end().

statusstrdefault: "running"

Starts as "running"; the local task-end hook sets "error" when result metadata contains a truthy error and "ok" otherwise.

Contents

metadatadict[str, Any]default: {}

Agent/task state, final state, retry count, budget decisions, and trace-level metric rollups.

spanslist[TraceSpan]default: []

Flat operation list in start order.

eventslist[TraceEvent]default: []

Chronological detailed inference events. Events associated with active spans also appear within the corresponding span.

TraceRecord.duration_ms and TraceRecord.to_dict​

property and methodprotolink.telemetry.TraceRecord inspection
source
duration_ms: float | None
to_dict() -> dict[str, Any]

Inspect total elapsed task time and serialize a replayable trace dictionary.

Returns

duration_msfloat | None

None until ended_at is set; otherwise elapsed milliseconds rounded to three decimals.

to_dict()dict[str, Any]

All trace fields plus computed duration, recursively serialized spans, and serialized trace-level events.

Raises

ValueError | AttributeError

Invalid timestamps or manually inserted span/event objects that do not satisfy the documented interface.

Intentional event duplication

An event recorded while a span is active is referenced by both TraceRecord.events and TraceSpan.events. Serialization emits it in both views so consumers can choose chronological replay or span-local inspection.

Hosted telemetry providers​

The hosted adapters import their SDKs lazily when constructed. Importing protolink.telemetry therefore does not itself require Langfuse or LangSmith. Both adapters isolate async task state with contextvars, catch ordinary provider-operation exceptions inside lifecycle hooks, and log warnings so export outages normally do not stop Agent work.

LangfuseTelemetry​

classprotolink.telemetry.LangfuseTelemetry
source
LangfuseTelemetry(
  public_key: str | None = None,
  secret_key: str | None = None,
  host: str | None = None,
) -> None

Create a Langfuse client and map Agent tasks to traces, complete inference cycles to generations, and explicit task-level tool calls to spans.

Parameters

public_keystr | Nonedefault: None

Langfuse public key. A truthy explicit value wins; otherwise the constructor reads LANGFUSE_PUBLIC_KEY.

secret_keystr | Nonedefault: None

Langfuse secret key. A truthy explicit value wins; otherwise the constructor reads LANGFUSE_SECRET_KEY.

hoststr | Nonedefault: None

Langfuse endpoint. Resolution is a truthy explicit value, then LANGFUSE_HOST, then "https://cloud.langfuse.com".

Attributes

langfuselangfuse.Langfuse

SDK client constructed immediately with the resolved credentials and host.

Raises

ImportError

The optional langfuse library is unavailable. Install langfuse directly or install protolink[telemetry].

Langfuse client error

Credential, host, configuration, or SDK-construction failures propagate from the constructor.

Falsey explicit values

Resolution uses Python's or. An empty explicit key or host does not override the environment/default; it falls through to the next source.

LangfuseTelemetry lifecycle methods​

async methodsprotolink.telemetry.LangfuseTelemetry lifecycle
source
async on_task_start(task: Task, agent_name: str) -> Any
async on_task_end(task: Task, result: Task, agent_name: str) -> Any
async on_llm_start(
  prompt: str,
  model: str | None = None,
  metadata: dict[str, Any] | None = None,
) -> Any
async on_llm_end(response: Part) -> Any
async on_tool_start(tool_name: str, args: dict[str, Any]) -> Any
async on_tool_end(
  tool_name: str,
  result: Any,
  error: str | None = None,
) -> Any
async on_llm_event(event: dict[str, Any]) -> Any

Translate the shared lifecycle into the Langfuse trace API. The six overridden methods return None implicitly. on_llm_event() is inherited from Telemetry and returns None without exporting its event.

Task mapping

on_task_start

Call langfuse.trace() with a "Task: "-prefixed agent name, the Task ID as the Langfuse trace ID, and agent-name metadata. It does not send the full task as trace input.

on_task_end

If a trace exists, update its output with result.to_dict(), flush the client, and clear the current trace in a finally block. The original task and agent_name parameters are not otherwise used.

LLM mapping

on_llm_start

If a trace exists, create a generation named "LLM Call" containing model, raw prompt input, and the supplied metadata. An empty or absent metadata mapping is sent as None.

on_llm_end

If a generation exists, end it with response.content; objects without that attribute fall back to str(response). Clear the current generation even when ending it fails.

Tool mapping

on_tool_start

If a trace exists, create a span with a "Tool: "-prefixed tool name and the arguments as input.

on_tool_end

End the active span with output=result when error is falsey. A truthy error instead ends it with level "ERROR" and status_message=error. Clear the current span afterward.

Fallback and errors

missing parent context

LLM/tool starts and all matching ends return silently when their required trace, generation, or span is absent.

provider operation failure

Every overridden hook catches Exception around SDK calls and logs a warning. End hooks still clear their corresponding context variable.

detailed inference events

The inherited no-op hook does not forward context_prepared, per-call metrics, retries, LLM-loop tools, delegation, or budget events.

Trace correlation

Langfuse uses task.id as its trace ID. It does not read the separate task.metadata["trace_id"] used by LocalTraceTelemetry, although that value is included in Agent-supplied LLM metadata when another tracker has already attached it.

LangSmithTelemetry​

classprotolink.telemetry.LangSmithTelemetry
source
LangSmithTelemetry(
  api_key: str | None = None,
  project_name: str | None = None,
) -> None

Create a LangSmith client and represent task execution as a root RunTree with child LLM and explicit tool runs.

Parameters

api_keystr | Nonedefault: None

LangSmith API key. A truthy explicit value wins; otherwise the constructor reads LANGCHAIN_API_KEY.

project_namestr | Nonedefault: None

Run project. Resolution is a truthy explicit value, then LANGCHAIN_PROJECT, then "default".

Attributes

clientlangsmith.Client

SDK client constructed immediately with the resolved API key.

project_namestr

Resolved project used for every root run.

Raises

ImportError

The optional langsmith package is unavailable. Install it directly or install protolink[telemetry].

LangSmith client error

SDK client construction and configuration errors propagate.

Falsey explicit values

As with Langfuse, empty explicit values fall through to environment/default values because resolution uses or.

LangSmithTelemetry lifecycle methods​

async methodsprotolink.telemetry.LangSmithTelemetry lifecycle
source
async on_task_start(task: Task, agent_name: str) -> Any
async on_task_end(task: Task, result: Task, agent_name: str) -> Any
async on_llm_start(
  prompt: str,
  model: str | None = None,
  metadata: dict[str, Any] | None = None,
) -> Any
async on_llm_end(response: Part) -> Any
async on_tool_start(tool_name: str, args: dict[str, Any]) -> Any
async on_tool_end(
  tool_name: str,
  result: Any,
  error: str | None = None,
) -> Any
async on_llm_event(event: dict[str, Any]) -> Any

Translate the common hooks into root and child LangSmith runs. The six overrides return None implicitly; detailed on_llm_event() values are ignored by the inherited base implementation.

Task mapping

on_task_start

Construct a RunTree with a "Task: "-prefixed agent name, run_type="chain", the configured project, a task-ID input mapping, agent-name metadata, and the shared client; post it and retain it as the current root.

on_task_end

End the active root with result.to_dict() as outputs, patch it to LangSmith, and clear context in finally. The output is passed directly rather than wrapped under a named key.

LLM mapping

on_llm_start

Create and post an llm child named "LLM Call" with prompt, model, and metadata in its inputs.

on_llm_end

End the active child with response content under the response output key, patch it, and clear context. Objects without content are stringified.

Tool mapping

on_tool_start

Create and post a tool child with a "Tool: "-prefixed tool name and the argument mapping as inputs.

on_tool_end

End with error=error for a truthy error, otherwise put the value under the result output key; patch and clear the current child.

Fallback and errors

missing parent context

Child starts and end hooks return silently when no corresponding current run exists.

provider operation failure

SDK construction/post/end/patch calls inside hook bodies catch Exception and log a warning. End hooks clear context even after an error.

dependency re-check

on_task_start() resolves RunTree through the lazy dependency helper before entering its SDK-operation try block. In the unusual case that the package becomes unavailable after construction, that ImportError propagates.

detailed inference events

Context, metric, retry, budget, LLM-loop tool, and delegation events are not exported by the inherited no-op hook.

Multiplexer API​

MultiTelemetry​

classprotolink.telemetry.MultiTelemetry
source
MultiTelemetry(
  trackers: list[Telemetry],
) -> None

Broadcast every telemetry hook to multiple trackers. This lets one Agent retain detailed local traces while also exporting the coarse lifecycle to one or more hosted systems.

Parameters

trackerslist[Telemetry]required

Ordered tracker list retained by reference. No copy, element validation, or non-empty check is performed; later list mutations affect future broadcasts.

Attributes

trackerslist[Telemetry]

Exact supplied list.

Useful ordering

Put LocalTraceTelemetry first when preserving an event locally is more important than reaching a later external exporter. This does not make delivery transactional, but it determines which trackers have already received a hook if a later tracker raises.

MultiTelemetry lifecycle methods​

async methodsprotolink.telemetry.MultiTelemetry lifecycle
source
async on_task_start(task: Task, agent_name: str) -> Any
async on_task_end(task: Task, result: Task, agent_name: str) -> Any
async on_llm_start(
  prompt: str,
  model: str | None = None,
  metadata: dict[str, Any] | None = None,
) -> Any
async on_llm_end(response: Part) -> Any
async on_tool_start(tool_name: str, args: dict[str, Any]) -> Any
async on_tool_end(
  tool_name: str,
  result: Any,
  error: str | None = None,
) -> Any
async on_llm_event(event: dict[str, Any]) -> Any

Forward each hook, with the same arguments, to every tracker in list order.

Task hook parameters

taskTaskrequired

Live task passed unchanged to each task-start or task-end hook.

resultTaskrequired

Task result passed unchanged to each task-end hook.

agent_namestrrequired

Agent name passed unchanged to each task hook.

LLM hook parameters

promptstrrequired

Inference prompt passed unchanged to each LLM-start hook.

modelstr | Nonedefault: None

Optional model identifier passed unchanged to each LLM-start hook.

metadatadict[str, Any] | Nonedefault: None

Optional mapping passed by reference to each LLM-start hook. A tracker that mutates it changes what later trackers observe.

responsePartrequired

Final inference Part passed unchanged to each LLM-end hook.

Tool hook parameters

tool_namestrrequired

Tool name passed unchanged to each explicit tool hook.

argsdict[str, Any]required

Mutable argument mapping passed by reference to each tool-start hook.

resultAnyrequired

Tool return value passed unchanged to each tool-end hook.

errorstr | Nonedefault: None

Optional failure text passed unchanged to each tool-end hook.

Detailed-event parameter

eventdict[str, Any]required

Event mapping passed by reference to each detailed-event hook. A tracker that mutates it changes what later trackers observe.

Returns

NoneNone

Tracker return values are discarded; after every await succeeds, the multiplexer returns None implicitly.

Raises

tracker error

Any exception propagates immediately. The failing tracker stops iteration, later trackers miss that hook, and no rollback is attempted for earlier trackers.

AttributeError

A list element does not implement the invoked hook.

Sequential fan-out

Trackers are awaited one at a time, not with asyncio.gather(). Ordering is deterministic, but total hook latency includes every tracker's latency. An empty list is valid and makes every hook a no-op.

Examples

from protolink import LocalTraceTelemetry
from protolink.telemetry import LangfuseTelemetry, MultiTelemetry

telemetry = MultiTelemetry(
    [
        LocalTraceTelemetry(path="traces.jsonl"),
        LangfuseTelemetry(),
    ]
)