Kotlin

Installazione

Compilato con Kotlin 2.4, per JVM 17 o successiva. Le uniche dipendenze sono kotlinx coroutines e serialization. L’HTTP lo fa il client della JDK, così non si aggiunge niente a un servizio che ha già un’idea sua sull’HTTP.

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

Costruire il client

Un’istanza per processo. Ogni campo di McpdConfig legge una variabile d’ambiente come proprio valore predefinito e si può impostare nel codice, così lo stesso codice gira in sviluppo e in produzione senza una modifica. Mcpd() senza argomenti funziona contro un mcpd locale.

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

Il codice vince sull’ambiente, e l’ambiente vince sull’impostazione predefinita.

enabled = false rende ogni chiamata di registrazione un’operazione a vuoto che restituisce comunque span validi, così mcpd si può spegnere nei test senza toccare il codice sotto test. Le letture vanno comunque al server: una query è una cosa che chi chiama sta aspettando, e una risposta vuota sarebbe una bugia.

Registrare un’esecuzione

span è la forma idiomatica. Chiude lo span, lo rende lo span attivo per tutto quello che il blocco chiama, e restituisce quello che il blocco ha restituito. Un blocco che solleva un’eccezione lascia uno span error che porta l’eccezione con sé, e la rilancia identica.

// Il valore del blocco è il valore di span, così la strumentazione non ti
// costringe a riscrivere il codice per farle posto.
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 blocco cancellato è un’altra cosa, e viene registrato in un altro modo. Una coroutine abbandonata da chi l’ha chiamata o da un timeout chiude lo span come cancelled anziché come error, perché un abbandono voluto non è un guasto dell’operazione e non deve leggersi come tale nell’elenco delle tracce.

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
}

Quando un blocco non ha la forma giusta, apri e chiudi lo span per conto tuo.

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

Contesto

Lo span attivo è un elemento del contesto della coroutine, quindi segue la coroutine e non il thread. Una funzione suspend parecchi livelli più in basso apre un figlio del padre giusto senza che nessun chiamante gliene passi uno, e due esecuzioni sullo stesso dispatcher non vedono mai gli span dell’altra. Lo leggi con mcpd.currentSpan(), e ne prendi l’header con mcpd.traceparent().

mcpd.span("support_agent", SpanKind.AGENT_RUN) { run ->
    // A draft non si passa niente. Trova l'esecuzione nel contesto della
    // coroutine, per quante chiamate suspend ci siano di mezzo.
    run.setOutput(draft(question))
}

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

Fra due servizi il contesto viaggia in un header traceparent W3C, che porta con sé la decisione di campionamento, così una traccia non è mai registrata a metà. Un header malformato fa partire una traccia nuova anziché far fallire la richiesta che lo portava.

// In uscita: passa l'header al servizio successivo.
request.header("traceparent", span.traceparent())

// In entrata: continua quella traccia invece di aprirne una nuova.
mcpd.span("checkout", traceparent = headers["traceparent"]) { span ->
    charge(span)
}

Modello, consumo ed errori

I conteggi dei token sono quelli che mcpd trasforma in costo all’ingest, quindi la cifra che vedi sulla traccia è la cifra che ti è stata fatturata. Registrare un errore porta nello stesso gesto lo stato dello span a error, e un messaggio c’è sempre: «qualcosa è fallito» non dice a nessuno che cosa fare.

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

// setError prende il throwable che hai già in mano, così il messaggio, la
// classe e lo stack vengono tutti da lì.
tool.setError(e, retryable = true)

// Oppure un messaggio diretto, che in ogni caso è obbligatorio.
tool.setError("the shipping service returned 503")

Gli altri sono setInput, setOutput, setMetadata, setAttribute, setAttributes, setTags, setCorrelation e addLink, più child per un figlio esplicito ed event per un punto nel tempo. Si possono chiamare tutti senza rischi da qualunque thread e nessuno di loro solleva eccezioni. Uno span che appartiene all’albero di un altro tracer lo nomina con externalParentSpanId, portato per la correlazione e mai usato come padre dentro l’albero di mcpd.

Payload, sink e span già conclusi

capturePayloads = false tiene input e output dentro il processo: vengono scartati prima che il record arrivi alla coda, così niente di sensibile esce con un flush, e un singolo span lo può sovrascrivere per sé. L’exporter consegna ogni batch a un Sink: il trasporto HTTP come impostazione predefinita, un RecordingSink che tiene i batch in memoria per un test, o un ConsoleSink che scrive una riga JSON per segnale. Un sink che solleva un’eccezione è un batch fallito, riferito una volta per intervallo e mai sollevato verso chi ha chiamato.

// Tieni i prompt e le completion dentro il processo: lo span porta comunque
// i suoi tempi, lo stato, il consumo, il modello, la correlazione e gli errori.
val quiet = Mcpd(McpdConfig(capturePayloads = false))
quiet.span("draft_answer", input = question, capturePayloads = true) { }  // questo invece si tiene

// Uno span già concluso da un altro tracer, messo in coda com'è: senza mai
// cambiargli il padre, i tempi o il campionamento. Un id malformato viene
// scartato e contato, mai sollevato.
mcpd.record(finishedSpan)

// Verifica che cosa è stato registrato, senza un 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()

Esecuzioni durevoli

Lo stesso client serve le esecuzioni durevoli del daemon. Workflow costruisce la definizione e conserva i corpi dei passi come lambda sospendibili; un Worker la registra, fa long polling per ottenere i lease e serve ognuno di essi su una coroutine, e startRun, watchRun, publishEvent e gli altri sono extension sul client che rispecchiano uno a uno la superficie delle esecuzioni. Un corpo solleva RetryableException per essere ritentato entro il budget del suo passo, e registra span sotto il proprio passo attraverso lo StepContext. Ogni span aperto attraverso il contesto somma il proprio consumo a quello del passo man mano che finisce, così un corpo che registra le proprie chiamate al modello non somma mai i token a mano; setUsage sovrascrive quello che gli span hanno sommato. Un Worker prende quattro hook facoltativi accanto a onError (onLease, onStepStart, onStepEnd, onConflict) e risponde a stats(): le prese in carico, i corpi in volo, i completamenti, i fallimenti per classe, i conflitti per motivo e i lease che non si sono potuti servire.

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

Che cosa fa ogni costrutto, il protocollo dei lease e l’API delle esecuzioni stanno su Workflow, Worker ed Esecuzioni.

Punteggi e feedback

min e max sono parametri obbligatori. Sette è ottimo su dieci e scarso su cento, quindi un numero senza una scala dichiarata non si può confrontare con niente. Il feedback è il giudizio di una persona e resta separato da un punteggio, perché i due rispondono a domande diverse e non vanno mai mediati insieme.

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

I log e le metriche si attaccano allo span attivo, così un picco su un grafico apre l’esecuzione che lo ha causato.

Flush e arresto

Registrare aggiunge a una coda in memoria con un tetto e ritorna; una coroutine worker raggruppa in batch, comprime e manda. flush() manda quello che è in coda e aspetta. shutdown() ferma il worker, manda quello che resta, e si può chiamare due volte senza rischi. L’export finale gira senza potere essere cancellato, perché all’arresto si arriva di solito da un finally durante una cancellazione, e l’ultimo batch è quello che spiega perché il processo se ne sta andando.

Mcpd è AutoCloseable, quindi funzionano sia use sia uno shutdown hook. Una funzione serverless o un worker di breve durata che non chiama né l’uno né l’altro perde il suo ultimo batch, perché il runtime smette di schedulare nel momento in cui l’handler ritorna.

// Un servizio JVM: arrestalo all'uscita.
Runtime.getRuntime().addShutdownHook(Thread { mcpd.close() })

// Un worker di breve durata: use, e si arresta alla parentesi che chiude.
Mcpd(McpdConfig(project = "batch")).use { mcpd ->
    mcpd.span("reindex") { run -> reindex(run) }
}

Che cosa ti torna indietro

I metodi di lettura rispecchiano uno a uno l’API HTTP e restituiscono i tipi generati. Sono gli unici metodi che sollevano eccezioni, e sollevano McpdException: una query che stai aspettando non è una cosa da inghiottire.

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

Su che cosa puoi contare

• La strumentazione non solleva mai eccezioni. Un export fallito, un server irraggiungibile o un valore che non si lascia serializzare vengono riferiti al massimo una volta ogni 30 secondi attraverso java.util.logging e per il resto inghiottiti. L’osservabilità non deve mai far cadere la cosa che osserva.
• Registrare non blocca, quindi la latenza del tuo servizio non dipende mai da quanto mcpd sia veloce, né dal fatto che sia attivo.
• La memoria è limitata. La coda tiene maxQueue segnali; oltre quel punto i più vecchi vengono scartati e contati, e si leggono in droppedSignals e in stats(). Un buffer senza limite trasforma un’interruzione di rete in un processo ucciso per memoria esaurita.
• La decisione di campionamento si prende una volta sola, alla radice, e si eredita.
• Uno span viene esportato quando parte e di nuovo quando finisce, così un’esecuzione andata in crash per strada resta visibile. Quando uno span finisce prima che il suo record di inizio sia stato mandato, il record viene aggiornato sul posto e ne parte uno solo.
• I batch vengono compressi con gzip sopra 1 KiB e portano una chiave di idempotenza che sopravvive a un nuovo tentativo, così una richiesta andata in timeout dopo che il server l’aveva archiviata non può contare due volte.
• Le durate si misurano su un orologio monotono, così uno scatto NTP non ne può produrre una negativa.

Trovare un’esecuzione copre i filtri per intero. Il modello dei dati elenca ogni campo che questi tipi portano.