Kotlin

Installation

Compilé avec Kotlin 2.4, pour une JVM 17 ou plus récente. Les seules dépendances sont kotlinx coroutines et serialization. Le HTTP passe par le client du JDK, si bien que rien n’est ajouté à un service qui a déjà son idée sur la question.

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

Construire le client

Une instance par processus. Chaque champ de McpdConfig lit une variable d’environnement comme valeur par défaut et se règle dans le code, si bien que le même code tourne en développement et en production sans la moindre modification. Mcpd() sans argument fonctionne face à un mcpd local.

// 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",
    )
)

Le code l’emporte sur l’environnement, et l’environnement l’emporte sur la valeur par défaut.

enabled = false fait de chaque appel d’enregistrement une opération sans effet qui renvoie quand même des spans valides : mcpd peut donc être coupé dans les tests sans toucher au code testé. Les lectures vont toujours au serveur : une requête est une chose que l’appelant attend, et une réponse vide serait un mensonge.

Enregistrer une exécution

span est la forme idiomatique. Elle termine le span, en fait le span actif pour tout ce que le bloc appelle, et renvoie ce que le bloc a renvoyé. Un bloc qui lève une exception laisse un span en error porteur de l’exception et la relance telle quelle.

// La valeur du bloc est celle de span, si bien que l’instrumentation ne vous
// oblige pas à restructurer le code autour d’elle.
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
}

Un bloc annulé est autre chose, et il est enregistré autrement. Une coroutine abandonnée par son appelant ou par un délai dépassé termine le span en cancelled plutôt qu’en error, car un abandon délibéré n’est pas un échec de l’opération et ne doit pas se lire comme tel dans la liste des traces.

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
}

Quand un bloc n’est pas la bonne forme, démarrez et terminez le span vous-même.

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

Contexte

Le span actif est un élément du contexte de coroutine : il suit donc la coroutine et non le thread. Une fonction suspend située plusieurs couches plus bas ouvre un enfant du bon parent sans qu’aucun appelant lui en passe un, et deux exécutions sur le même dispatcher ne voient jamais les spans l’une de l’autre. Lisez-le avec mcpd.currentSpan(), et obtenez son en-tête avec mcpd.traceparent().

mcpd.span("support_agent", SpanKind.AGENT_RUN) { run ->
    // draft ne reçoit rien. Il retrouve l’exécution par le contexte de
    // coroutine, quel que soit le nombre d’appels suspend franchis.
    run.setOutput(draft(question))
}

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

Entre services, le contexte voyage dans un en-tête W3C traceparent, qui emporte avec lui la décision d’échantillonnage : une trace n’est donc jamais enregistrée à moitié. Un en-tête mal formé démarre une nouvelle trace plutôt que de faire échouer la requête qui le transportait.

// En sortie, passez l’en-tête au service suivant.
request.header("traceparent", span.traceparent())

// En entrée, continuez cette trace au lieu d’en démarrer une nouvelle.
mcpd.span("checkout", traceparent = headers["traceparent"]) { span ->
    charge(span)
}

Modèle, consommation et erreurs

mcpd convertit les décomptes de tokens en coût à l’ingestion : le montant porté par la trace est donc celui qui vous a été facturé. Enregistrer une erreur place du même geste le statut du span sur error, et un message est toujours présent : « quelque chose a échoué » ne dit pas quoi faire.

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

// setError prend le throwable que vous détenez déjà, si bien que le message,
// la classe et la pile en sont tous tirés.
tool.setError(e, retryable = true)

// Ou un message directement, qui est de toute façon obligatoire.
tool.setError("the shipping service returned 503")

Les autres sont setInput, setOutput, setMetadata, setAttribute, setAttributes, setTags, setCorrelation et addLink, plus child pour un enfant explicite et event pour un point dans le temps. Chacune peut être appelée depuis n’importe quel thread et aucune ne lève d’exception. Un span qui appartient à l’arbre d’un autre traceur le nomme avec externalParentSpanId, porté pour la corrélation et jamais utilisé comme parent dans l’arbre propre à mcpd.

Payloads, sinks et spans terminés

capturePayloads = false garde input et output dans le processus : ils sont écartés avant que l’enregistrement n’atteigne la file d’attente, si bien que rien de sensible ne sort lors d’un flush, et un span peut décider autrement pour lui-même. L’exporter remet chaque lot à un Sink : le transport HTTP par défaut, un RecordingSink qui garde les lots en mémoire pour un test, ou un ConsoleSink qui écrit une ligne JSON par signal. Un sink qui lève une exception donne un lot en échec, signalé une fois par intervalle et jamais relancé chez l’appelant.

// Garder les prompts et les complétions dans le processus. Le span porte
// toujours ses temps, son statut, sa consommation, son modèle, sa corrélation
// et ses erreurs.
val quiet = Mcpd(McpdConfig(capturePayloads = false))
quiet.span("draft_answer", input = question, capturePayloads = true) { }  // celui-ci, conservé

// Un span qu’un autre traceur a déjà terminé, mis en file tel quel, jamais
// re-parenté, re-daté ni ré-échantillonné. Un identifiant mal formé est écarté
// et compté, jamais levé.
mcpd.record(finishedSpan)

// Prouver ce qui a été enregistré, sans serveur.
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()

Exécutions durables

Le même client dessert les exécutions durables du daemon. Workflow construit la définition et retient le corps des étapes sous forme de lambdas suspendues ; un Worker l’enregistre, attend des leases en long polling et sert chacun d’eux sur une coroutine, et startRun, watchRun, publishEvent et les autres sont des extensions du client qui reprennent un pour un la surface des exécutions. Un corps d’étape lève RetryableException pour être réessayé dans le budget de l’étape, et enregistre des spans sous son étape via le StepContext. Chaque span ouvert via le contexte ajoute sa consommation à celle de l’étape quand il se termine : un corps qui enregistre ses appels de modèle ne totalise donc jamais les tokens à la main ; setUsage remplace ce que les spans ont additionné. Un Worker accepte quatre hooks facultatifs à côté de onError (onLease, onStepStart, onStepEnd, onConflict) et répond à stats() : leases prises, corps en cours, achèvements, échecs par classe, conflits par motif et leases non servables.

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}"))

Ce que fait chaque construction, le protocole de lease et l’API des exécutions sont décrits dans Workflows, Workers et Exécutions.

Scores et feedback

min et max sont des paramètres obligatoires. Sept, c’est excellent sur dix et médiocre sur cent : un nombre sans échelle déclarée ne peut donc être comparé à rien. Le feedback est le jugement d’une personne et reste distinct d’un score, car les deux répondent à des questions différentes et ne doivent jamais être moyennés ensemble.

// 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()

Les logs et les métriques se rattachent au span actif, si bien qu’un pic sur un graphique ouvre l’exécution qui l’a causé.

Flush et arrêt

Enregistrer ajoute à une file d’attente bornée en mémoire et rend la main ; une coroutine worker regroupe, compresse et envoie. flush() envoie ce qui est en file et attend. shutdown() arrête le worker, envoie ce qui reste, et peut être appelé deux fois sans risque. L’export final s’exécute sans pouvoir être annulé, car on arrive généralement à l’arrêt depuis un finally pendant une annulation, et le dernier lot est celui qui explique pourquoi le processus s’en va.

Mcpd est AutoCloseable : use et un hook d’arrêt fonctionnent donc tous les deux. Une fonction serverless ou un worker de courte durée qui n’appelle ni l’un ni l’autre perd son dernier lot, car le runtime cesse d’ordonnancer dès que le handler rend la main.

// Un service JVM, arrêté en sortant.
Runtime.getRuntime().addShutdownHook(Thread { mcpd.close() })

// Un worker de courte durée. Avec use, il s’arrête à l’accolade fermante.
Mcpd(McpdConfig(project = "batch")).use { mcpd ->
    mcpd.span("reindex") { run -> reindex(run) }
}

Ce que vous récupérez

Les méthodes de lecture reflètent l’API HTTP une pour une et renvoient les types générés. Ce sont les seules méthodes qui lèvent une exception, et elles lèvent McpdException : une requête que vous attendez n’est pas quelque chose à avaler.

$ 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",
)

Ce sur quoi vous pouvez compter

• L’instrumentation ne lève jamais d’exception. Un export en échec, un serveur injoignable ou une valeur qui refuse de se sérialiser sont signalés au plus une fois toutes les 30 secondes via java.util.logging, et avalés le reste du temps. L’observabilité ne doit jamais faire tomber ce qu’elle observe.
• L’enregistrement ne bloque pas : la latence de votre service ne dépend donc jamais de la vitesse de mcpd, ni de sa disponibilité.
• La mémoire est bornée. La file contient maxQueue signaux ; au-delà, les plus anciens sont écartés et comptés, lisibles dans droppedSignals et dans stats(). Un tampon sans limite transforme une panne réseau en arrêt pour mémoire épuisée.
• La décision d’échantillonnage est prise une seule fois, à la racine, puis héritée.
• Un span est exporté quand il démarre et de nouveau quand il se termine, si bien qu’une exécution qui a planté en plein vol reste visible. Quand un span se termine avant que son enregistrement de début ne soit parti, l’enregistrement est mis à jour sur place et un seul est envoyé.
• Les lots sont compressés en gzip au-delà de 1 KiB et portent une clé d’idempotence qui survit à un nouvel essai, si bien qu’une requête ayant expiré après que le serveur l’a stockée ne peut pas compter deux fois.
• Les durées sont mesurées sur une horloge monotone : un saut NTP ne peut donc pas en produire une négative.

Retrouver une exécution couvre les filtres en détail. Le modèle de données liste chaque champ que ces types portent.