Kotlin

Install

Built with Kotlin 2.4, for JVM 17 or later. The only dependencies are kotlinx coroutines and serialization. HTTP is the JDK's own client, so nothing is added to a service that already has an opinion about HTTP.

dependencies {
    implementation("dev.mcpd:mcpd-kotlin:0.1.0")
}

Construct the client

One instance per process. Every field of McpdConfig reads an environment variable as its default and is settable in code, so the same code runs in development and in production without an edit. Mcpd() with no arguments works against a local mcpd.

// Every field also reads an environment variable (MCPD_ENDPOINT,
// MCPD_PROJECT, MCPD_API_KEY, ...), so `Mcpd()` on its own works too.
val mcpd = Mcpd(
    McpdConfig(
        endpoint = "http://localhost:4317",
        project = "support-desk",
        environment = "development",
        serviceName = "support-agent",
        serviceVersion = "1.4.0",
    )
)

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 spans, so mcpd can be turned off in tests without touching the code under test. Reads still go to the server: a query is something the caller is waiting on, and an empty answer would be a lie.

Record a run

span is the idiomatic form. It ends the span, makes it the active span for everything the block calls, and returns whatever the block returned. A block that throws leaves an error span carrying the exception and rethrows it unchanged.

// The block's value is the value of span, so instrumentation does not force
// you to restructure the code around it.
val answer = mcpd.span("retrieve_context", SpanKind.RETRIEVAL) { span ->
    span.setInput(query)
    val docs = index.search(query)
    span.setOutput(docs.map { it.id })
    span.setAttribute("retrieval.hits", docs.size)
    docs
}

A cancelled block is different, and is recorded differently. A coroutine abandoned by its caller or by a timeout ends the span as cancelled rather than error, because a deliberate abandonment is not a failure of the operation and must not read as one in the trace list.

val question = "where is my order 41?"

// `span` starts the operation, ends it, and records anything it throws.
// Inside the block the span is the active one, so the helpers below find
// their parent through the coroutine context rather than being handed it.
val traceId = mcpd.span(
    "support_agent",
    SpanKind.AGENT_RUN,
    input = question,
    sessionId = "session-8f21",
    userId = "customer-4477",
    tags = listOf("support", "email"),
) { run ->
    mcpd.log(LogLevel.INFO, "handling a question", mapOf("channel" to "email"))

    val plan = planAnswer(mcpd, question)
    val order = lookUpOrder(mcpd, 41)
    val estimate = estimateDelivery(mcpd, 41)

    val answer = buildString {
        append(plan).append(' ').append(order)
        if (estimate != null) append(" It should arrive ").append(estimate).append('.')
    }
    run.setOutput(answer)
    mcpd.metric("support.answers", 1.0, MetricKind.COUNTER, unit = "{answer}")

    // The block's value is the value of `span`.
    run.traceId
}

When a block is not the right shape, start and end the span yourself.

val run = mcpd.trace("nightly_reindex", SpanKind.WORKFLOW_RUN, tags = listOf("batch"))
try {
    reindex()
    run.end()
} catch (e: Exception) {
    run.setError(e)
    run.end()
}

Context

The active span is an element of the coroutine context, so it follows the coroutine rather than the thread. A suspend function several layers down opens a child of the right parent without any caller passing it one, and two runs on the same dispatcher never see each other's spans. Read it with mcpd.currentSpan(), and get its header with mcpd.traceparent().

mcpd.span("support_agent", SpanKind.AGENT_RUN) { run ->
    // draft is handed nothing. It finds the run through the coroutine context,
    // however many suspend calls down it is.
    run.setOutput(draft(question))
}

suspend fun draft(question: String): String =
    mcpd.span("draft_answer", SpanKind.MODEL_GENERATION) { generation ->
        generation.setInput(question)
        "..."
    }

Between services, context travels in a W3C traceparent header, which carries the sampling decision with it, so a trace is never half recorded. A malformed header starts a new trace rather than failing the request that carried it.

// Outgoing: hand the header to the next service.
request.header("traceparent", span.traceparent())

// Incoming: continue that trace instead of starting a new one.
mcpd.span("checkout", traceparent = headers["traceparent"]) { span ->
    charge(span)
}

Model, usage and errors

Token counts are what mcpd turns into cost at ingest, so the figure on the trace is the figure you were billed. Recording an error sets the span's status to error in the same move, and a message is always present: "something failed" is not actionable.

setModelParameters records the settings the call was made with: temperature, topP, topK, maxTokens, seed, stopSequences and hasResponseFormat. They are settings rather than payload, so they survive capturePayloads = 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 hasResponseFormat is not recorded, only whether one was in force.

generation.setModel("gpt-4o", provider = "openai")
generation.setUsage(inputTokens = 1_180, outputTokens = 96)

// How the model was called. Settings, not payload: kept with capture off.
generation.setModelParameters(temperature = 0.0, seed = 0, maxTokens = 1_024)

// setError takes the throwable you are already holding, so the message, the
// class and the stack all come off it.
tool.setError(e, retryable = true)

// Or a message directly, which is required either way.
tool.setError("the shipping service returned 503")

The rest are setInput, setOutput, setMetadata, setAttribute, setAttributes, setTags, setCorrelation and addLink, plus child for an explicit child and event for a point in time. Every one is safe to call from any thread and none of them throw. A span that belongs to another tracer's tree names it with externalParentSpanId, carried for correlation and never used as a parent inside mcpd's own tree.

Payloads, sinks and finished spans

capturePayloads = false keeps input and output in the process: they are dropped before the record reaches the queue, so nothing sensitive leaves on a flush, and one span can override it for itself. The exporter hands every batch to a Sink: the HTTP transport by default, a RecordingSink that keeps batches in memory for a test, or a ConsoleSink that writes one JSON line per signal. A sink that throws is a failed batch, reported once per interval and never raised at the caller.

// Keep prompts and completions in the process: the span still carries its
// timing, status, usage, model, model parameters, correlation and errors.
val quiet = Mcpd(McpdConfig(capturePayloads = false))
quiet.span("draft_answer", input = question, capturePayloads = true) { }  // this one, kept

// A span another tracer already finished, queued as given: never re-parented,
// re-timed or re-sampled. A malformed id is dropped and counted, never thrown.
mcpd.record(finishedSpan)

// Prove what was recorded without a server.
val recorder = RecordingSink()
val local = Mcpd(McpdConfig(sink = recorder))
local.span("support_agent") { run -> run.setUsage(inputTokens = 12, outputTokens = 3) }
local.flush()
check(recorder.spans.single().usage?.inputTokens == 12L)

// queued, dropped, capacity, exported, failed, rejected.
val stats = local.stats()

Durable runs

The same client serves the daemon's durable runs. Workflow builds the definition and keeps the step bodies as suspending lambdas; a Worker registers it, long-polls for leases and serves each one on a coroutine, and startRun, watchRun, publishEvent and the rest are extensions on the client that mirror the run surface one for one. A body throws RetryableException to be retried within the step's budget, and records spans under its step through the StepContext. Every span opened through the context adds its usage to the step's as it ends, so a body that records its model calls never sums tokens by hand; setUsage overrides what the spans summed. A Worker takes four optional hooks beside onError (onLease, onStepStart, onStepEnd, onConflict) and answers stats(): claims, bodies in flight, completions, failures by class, conflicts by reason and unservable leases.

val support = Workflow("support_ticket")
    .step("classify") { ctx ->
        val ticket = ctx.input<JsonObject>()["ticket"]!!.jsonPrimitive.content
        mapOf("kind" to if (ticket.startsWith("Q")) "question" else "bug")
    }
    .agent("draft", StepOptions(input = mapOf("kind" to "\${steps.classify.output.kind}"))) {
        ctx ->
        // Your own model call, recorded under this step in the run's trace.
        // Its usage is summed into the step, so the daemon prices the run.
        ctx.span("model_call", SpanKind.MODEL_GENERATION) { span ->
            span.setModel("claude-sonnet-4", "anthropic")
            span.setUsage(inputTokens = 812, outputTokens = 37)
        }
        "an answer to a " + ctx.input<JsonObject>()["kind"]!!.jsonPrimitive.content
    }
    .waitFor(
        "approval", "approved", "\${input.ticket}", 7.days,
        StepOptions(guard = predicate("steps.classify.output.kind", "eq", "bug")),
    )
    .step(
        "send",
        StepOptions(
            input = mapOf("text" to "\${steps.draft.output}"),
            effect = Effect.AT_MOST_ONCE,
            retry = RetryPolicy(
                maxAttempts = 3, backoff = Backoff.EXPONENTIAL, initial = 500.milliseconds,
            ),
        ),
    ) { ctx ->
        val text = ctx.input<JsonObject>()["text"]!!.jsonPrimitive.content
        if (!deliver(text, ctx.idempotencyKey)) {
            throw RetryableException("mail relay busy", retryAfter = 2.seconds)
        }
        mapOf("sent" to true)
    }
    .output(mapOf("sent" to "\${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 required parameters. Seven is excellent out of ten and poor out of a hundred, so a number with no declared scale cannot be compared to anything. Feedback is a person's judgement and stays separate from a score, because the two answer different questions and must never be averaged together.

// A machine evaluation. `min` and `max` are required: 7 is excellent out of
// 10 and poor out of 100, and a score with no scale cannot be compared.
mcpd.score(
    traceId = traceId,
    scorerId = "answer_relevance",
    value = 0.86,
    min = 0.0,
    max = 1.0,
    reason = "answered the question asked, without the delivery estimate",
)

// A person's judgement, from the thumbs-up control in your own product.
mcpd.feedback(
    traceId = traceId,
    kind = FeedbackKind.THUMBS,
    value = "up",
    comment = "fast and clear",
    authorId = "customer-4477",
)

// Nothing above touched the network. This is where it goes.
mcpd.flush()

Logs and metrics attach to the active span, so a spike on a chart opens the run that caused it.

Flush and shutdown

Recording appends to a bounded in-memory queue and returns; a worker coroutine batches, compresses and sends. flush() sends what is queued and waits. shutdown() stops the worker, sends what is left, and is safe to call twice. The final export runs uncancellable, because shutdown is usually reached from a finally during cancellation and the last batch is the one that explains why the process is going away.

Mcpd is AutoCloseable, so use and a shutdown hook both work. A serverless function or a short-lived worker that calls neither loses its last batch, because the runtime stops scheduling the moment the handler returns.

// A JVM service: shut down on the way out.
Runtime.getRuntime().addShutdownHook(Thread { mcpd.close() })

// A short-lived worker: use, and it shuts down at the closing brace.
Mcpd(McpdConfig(project = "batch")).use { mcpd ->
    mcpd.span("reindex") { run -> reindex(run) }
}

What you get back

The read methods mirror the HTTP API one for one and return the generated types. They are the only methods that throw, and they throw McpdException: a query you are waiting on is not something to swallow.

$ gradle run> Task :runsupport_agent: 4 spans, ok  agent_run support_agent 79ms  model_generation plan_answer 41ms  tool_call lookup_order 16ms  tool_call estimate_delivery 11ms shipping service returned 503 for order 41

val detail = mcpd.getTrace(traceId, include = listOf("scores", "feedback"))
println("${detail.trace.name}: ${detail.spans.size} spans, ${detail.trace.status}")
for (span in detail.spans) {
    val ms = (span.durationNs ?: 0) / 1_000_000
    println("  ${span.kind} ${span.name} ${ms}ms ${span.error?.message ?: ""}")
}

val summary = mcpd.analyticsSummary(
    from = Clock.System.now() - 1.hours,
    to = Clock.System.now(),
)
println(
    "last hour: ${summary.traceCount} runs, " +
        "${summary.errorCount} failed, ${summary.totalCost} USD",
)

What you can rely on

• Instrumentation never throws. A failed export, an unreachable server or a value that will not serialise is reported at most once every 30 seconds through java.util.logging and otherwise swallowed. Observability must never take down the thing it observes.
• Recording does not block, so your service's latency never depends on mcpd being fast, or up.
• Memory is bounded. The queue holds maxQueue signals; past that the oldest are dropped and counted, readable as droppedSignals and in stats(). An unbounded buffer turns a network outage into an out-of-memory kill.
• The sampling decision is made once, at the root, and inherited.
• A span is exported when it starts and again when it ends, so a run that crashed mid-flight is still visible. When a span ends before its start record has been sent, the record is updated in place and only one is sent.
• Batches are gzipped above 1 KiB and carry an idempotency key that survives a retry, so a request that timed out after the server stored it cannot double-count.
• Durations are measured from a monotonic clock, so an NTP step cannot produce a negative one.

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