Python

Install

Python 3.9 or newer. There are no required dependencies, deliberately: the exporter runs on urllib and gzip from the standard library, so installing mcpd adds nothing to a lockfile and cannot conflict with what your agent already depends on.

The httpx extra adds connection reuse and nothing else. The SDK behaves identically without it.

Construct the client

One client per process. Every argument also reads an environment variable, so the same code runs in development and in production without an edit. Mcpd() with no arguments talks to a daemon on http://localhost:4317.

from mcpd import FeedbackKind, Mcpd, SpanKind

# Everything here has an MCPD_* environment variable behind it, so the same code
# runs in development and in production with no edit. `Mcpd()` with no arguments
# talks to a local daemon.
mcpd = Mcpd(
    endpoint=os.environ.get("MCPD_ENDPOINT", "http://localhost:4317"),
    project="support",
    environment="development",
    service_name="support-agent",
    service_version="1.4.2",
)

Code wins over the environment, and the environment wins over the default.

enabled=False makes every recording call a no-op that still returns valid objects, so you can disable mcpd in tests without changing the code under test. capture_payloads=False keeps prompts and completions in the process: spans are still recorded with their timing, status, usage, model, model parameters, correlation and errors, and input and output are dropped before they reach the queue. A span can say otherwise for itself with capture_payloads=True in its start fields, and the other way round.

Record a run

mcpd.trace(name) starts a run and mcpd.span(name) starts a step under whatever is currently active. Both are context managers: the block ends the span on the way out and records the exception if it raised, then lets that exception through unchanged. Your own error handling is not altered by being observed.

# The root span is the run. Everything opened inside it becomes a child,
# because the active span lives in a `contextvars.ContextVar`.
with mcpd.trace("support_agent", session_id="sess-8121", user_id=user_id,
                tags=["support", "orders"]) as run:
    run.set_input({"question": question})

    # A model call. `set_usage` is what mcpd prices the run from, so the
    # cost on the trace is the cost you were actually billed.
    with mcpd.span("plan", kind=SpanKind.MODEL_GENERATION) as generation:
        generation.set_model("gpt-4o-mini", provider="openai")
        generation.set_input([
            {"role": "system", "content": "You are a support agent."},
            {"role": "user", "content": question},
        ])
        time.sleep(0.12)  # stand-in for the provider round trip
        generation.set_output("I should look the order up first.")
        generation.set_usage(input_tokens=812, output_tokens=96, cache_read=128)

    # A tool call that fails. The `with` block records the exception on the
    # span, marks it as an error, and lets the exception through unchanged.
    order = None
    try:
        with mcpd.span("lookup_order", kind=SpanKind.TOOL_CALL) as tool:
            tool.set_entity("tool", entity_id="orders.lookup", name="Order lookup")
            tool.set_input({"order_id": "A-7741"})
            order = lookup_order("A-7741")
            tool.set_output(order)
    except TimeoutError as exc:
        # The span already carries the error. This is the agent deciding
        # what to do about it, which is a different question.
        mcpd.log("warn", "order lookup failed, answering from context", {"error": str(exc)})
        mcpd.metric("support.tool_failures", 1, "counter", labels={"tool": "orders.lookup"})

    reply = ("I could not reach the order system just now. Your order A-7741 "
             "shipped on Tuesday and tracking should update within the hour.")
    run.set_output(reply)
    run.set_metadata({"used_fallback": order is None})
    return reply, run.trace_id

Without a with block, call run.end() yourself. The span is exported when it starts either way, so a span that never ends is visible as a span that never ended, which is usually the one you are looking for.

# Sync and async are the same object. Both end the span on the way out and
# record the exception if the block raised.
with mcpd.trace("nightly_summariser") as run:
    with mcpd.span("fetch", kind=SpanKind.TOOL_CALL) as step:
        step.set_output({"documents": 42})

async with mcpd.trace("agent") as run:
    async with mcpd.span("generate", kind=SpanKind.MODEL_GENERATION) as generation:
        generation.set_output(await model.complete(prompt))
    await mcpd.aflush()

Context

The active span is a contextvars.ContextVar. It follows an await into a task and does not leak between threads, so a helper three frames down opens a child of the right parent without any caller threading an argument through it.

def draft(question: str) -> str:
    # Nothing was passed in. The span still lands under the run, because the
    # active span is a contextvars.ContextVar and this call is inside the block.
    with mcpd.span("draft", kind=SpanKind.MODEL_GENERATION) as generation:
        generation.set_input(question)
        return "..."


with mcpd.trace("support_agent") as run:
    run.set_output(draft("where is my order?"))

What comes out is the tree, with the timings and the payloads attached.

Between services, context travels in a W3C traceparent header. A malformed header starts a new trace rather than failing the request that carried it. An inbound decision not to sample is obeyed, because a trace that some services keep and others drop is not a trace.

# Outbound: hand the active span to the next service.
requests.post(url, headers={"traceparent": mcpd.traceparent()}, json=payload)

# Inbound: continue that trace rather than starting a new one.
with mcpd.trace("handler", traceparent=request.headers.get("traceparent")) as run:
    ...

Model, usage and errors

Token counts are what mcpd prices at ingest, so the cost on the trace is the cost you were billed. Report what the provider told you. set_error accepts a string or an exception, sets status to error in the same move, and always populates a message: "something failed" is not actionable.

set_model_parameters records the settings the call was made with: temperature, top_p, top_k, max_tokens, seed, stop_sequences and has_response_format. They are settings rather than payload, so they survive capture_payloads=False, which is the case that matters: they are what somebody reproducing the run has to match. An argument you leave out is absent and one you pass is kept whatever it is, so seed=0 is a seed. The schema behind has_response_format is not recorded, only whether one was in force.

generation.set_model("gpt-4o-mini", provider="openai")
generation.set_usage(input_tokens=812, output_tokens=96, cache_read=128)

# How the model was called. Settings, not payload: kept with capture off.
generation.set_model_parameters(temperature=0, seed=0, max_tokens=1024)

# A failure you caught and decided about yourself.
generation.set_error("the provider timed out", kind="TimeoutError", retryable=True)

# A point in time inside the span, with no duration of its own.
generation.event("first_token")

The other setters are set_input, set_output, set_metadata, set_attributes, set_tags, set_entity, set_cost, set_status and add_link. All of them return the span, and any of them can also be passed as a keyword when the span starts. external_parent_span_id= names a parent from a tracing system that is not mcpd, carried for correlation and never used for parenting.

Payloads, sinks and finished spans

A sink is where a drained batch goes. The HTTP transport is the default. A RecordingSink keeps every batch in memory, so a test asserts what the instrumentation recorded without a server; a ConsoleSink prints one JSON line per signal to stderr. A sink that raises is a failed batch, counted and reported, never an exception in your code. stats() answers what the client has done since it was built.

from mcpd import Mcpd, RecordingSink

sink = RecordingSink()
mcpd = Mcpd(sink=sink)
with mcpd.trace("answer") as run:
    run.set_usage(input_tokens=12)
mcpd.flush()
assert sink.spans[0].usage.input_tokens == 12
print(mcpd.stats())  # queued, dropped, capacity, exported, failed, rejected

mcpd.record(span) queues a SpanRecord that already ended, ids, parent and times as you hold them: what an adapter for another tracing system or a batch import has. It is never re-parented, re-timed or re-sampled, and a malformed record is dropped and counted in stats().rejected, never raised.

Durable runs

The same client serves the daemon's durable runs. workflow() builds the definition and keeps the step bodies, as calls or as decorators; a Worker registers it, long-polls for leases on a thread pool and serves each one, and mcpd.runs, mcpd.events and mcpd.definitions mirror the run surface one for one. A body raises StepFailure with class_="retryable" to be retried within the step's budget, and records spans under its step through the StepContext. What those spans spend is the step's usage: tokens are summed as each span ends and the last model seen wins, so the daemon prices the step from what the trace shows; ctx.set_usage overrides it field by field. Worker takes four hooks beside on_error (on_lease, on_step_start, on_step_end, on_conflict), and worker.stats() counts leases claimed, bodies in flight, completed, failed by class, conflicts by reason and unservable leases.

support = workflow("support_ticket")


@support.step("classify")
def classify(ctx, input, prior):
    return {"kind": "question" if input["ticket"].startswith("Q") else "bug"}


@support.agent("draft", input={"kind": "${steps.classify.output.kind}"})
def draft(ctx, input, prior):
    # Your own model call, recorded under this step in the run's trace. What
    # the span spends is the step's usage: the daemon prices it from here.
    with ctx.span("model_call", kind="model_generation") as span:
        span.set_model("claude-sonnet-4", "anthropic")
        span.set_usage(input_tokens=812, output_tokens=37)
        text = f"an answer to a {input['kind']}"
    return text


support.wait_for("approval", event="approved", key="${input.ticket}",
                 timeout_ms=7 * 24 * 3_600_000,
                 when=when("steps.classify.output.kind", "eq", "bug"))


@support.step("send", input={"text": "${steps.draft.output}"}, effect="at_most_once",
              retry=RetryPolicy(max_attempts=3, backoff="exponential", initial_ms=500))
def send(ctx, input, prior):
    if not deliver(input["text"], ctx.idempotency_key):
        raise StepFailure("mail relay busy", class_="retryable", retry_after_ms=2_000)
    return {"sent": True}


support.output({"sent": "${steps.send.output.sent}"})

What each construct does, the lease protocol and the run API are on Workflows, Workers and Runs.

Scores and feedback

min and max are keyword-only and have no defaults. A bare number with no declared scale cannot be compared to anything, so the scale is enforced at the call site rather than guessed later. Feedback is a person's judgement and stays separate from a score: the two answer different questions and must never be averaged together.

# A machine evaluation. `min` and `max` are required: a bare number with no
# declared scale cannot be compared to anything.
mcpd.score(
    "answer_relevance",
    value=round(random.uniform(0.6, 0.95), 2),
    min=0.0,
    max=1.0,
    trace_id=trace_id,
    reason="named the order and gave a delivery window, without the live lookup",
)

# A human evaluation. Kept separate from a score, because a person's
# judgement and a model's must never be averaged together.
mcpd.feedback(
    FeedbackKind.THUMBS,
    "up",
    trace_id=trace_id,
    comment="fine given the outage",
    author_kind="end_user",
)

Both attach to the active span's trace when you do not name one. Logs and metrics do the same, so a spike on a chart opens the run that caused it.

Flush and shutdown

Recording appends to an in-memory queue and returns; a daemon thread does the HTTP. flush() sends what is waiting and returns False if it could not do so within its timeout. shutdown() flushes, stops the worker and releases the transport. Both are idempotent. Inside an event loop call aflush() and ashutdown(), which do the same work without blocking the loop.

In a serverless function this matters more than anywhere else. The runtime freezes the process the moment your handler returns, and the worker thread never runs again, so a batch that was still queued is lost.

def handler(event, context):
    with mcpd.trace("answer") as run:
        run.set_input(event["body"])
        answer = agent(event["body"])
        run.set_output(answer)

    # The batch is still in memory here. The runtime freezes this process the
    # moment the handler returns, and the worker thread never runs again.
    mcpd.flush()
    return {"statusCode": 200, "body": answer}

A client also registers an atexit hook, so an ordinary script that forgets to shut down still gets its last batch. That hook does not run on os._exit() or on a fatal signal, so it is a safety net and not a replacement.

What you can rely on

• Instrumentation never raises. Every recording call is guarded; a failure is logged at most once per interval and swallowed. Observability must never take down the thing it observes. Read calls are the exception and do raise, because a query you are waiting on is different.
• Recording does not block. Your agent's latency never depends on mcpd being fast, or up.
• Memory is bounded. max_queue signals, 8192 by default. When it is full the oldest are dropped, mcpd.dropped counts them and one warning is logged per interval. An unbounded buffer turns a network outage into an out-of-memory kill.
• The sampling decision is made once, at the root, from the trace id, and inherited by every child. An inbound traceparent always wins.
• A span is exported when it starts and updated when it ends, so a run that crashes mid-flight is still visible.
• Timestamps come from time.time_ns() and go on the wire as UTC with nine digits. Local time makes a distributed trace unreadable.

Reading it back

The read methods mirror the HTTP API and return the generated types. They are synchronous and raise McpdError on failure. from_ and to accept a datetime or a shorthand such as -15m, -1h or -7d. In async code, run them off the loop with asyncio.to_thread.

# Nothing has been sent yet: recording appends to an in-memory queue and a
# background worker does the I/O. Flush before you read your own writes.
mcpd.flush()

detail = mcpd.get_trace(trace_id, include=["logs", "scores", "feedback"])
print(f"\n{detail.trace.name}  {detail.trace.duration_ns / 1e6:.0f} ms  "
      f"{detail.trace.span_count} spans  {detail.trace.error_count} errors")
for span in detail.spans:
    cost = f"  ${span.cost.total_cost:.6f}" if span.cost else ""
    print(f"  {span.kind.value:<18} {span.name:<14} {span.status.value}{cost}")

summary = mcpd.analytics_summary("-1h", "-0s")
print(f"\nlast hour  traces={summary.trace_count}  errors={summary.error_count}  "
      f"p95={summary.latency.p95:.0f}ms  cost=${summary.total_cost:.4f}")

Finding a run covers the filters in full. The data model lists every field these types carry.