Kotlin

Instalación

Compilado con Kotlin 2.4, para JVM 17 o posterior. Las únicas dependencias son kotlinx coroutines y serialization. El HTTP lo hace el cliente propio del JDK, así que no se añade nada a un servicio que ya tiene su propio criterio sobre HTTP.

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

Construir el cliente

Una instancia por proceso. Cada campo de McpdConfig lee una variable de entorno como valor por defecto y se puede asignar en código, así que el mismo código corre en desarrollo y en producción sin ninguna edición. Mcpd() sin argumentos funciona contra 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",
    )
)

El código gana al entorno, y el entorno gana al valor por defecto.

enabled = false convierte cada llamada de registro en una operación vacía que sigue devolviendo spans válidos, así que se puede apagar mcpd en las pruebas sin tocar el código bajo prueba. Las lecturas siguen yendo al servidor: una consulta es algo que quien llama está esperando, y una respuesta vacía sería mentira.

Registrar una ejecución

span es la forma idiomática. Cierra el span, lo convierte en el span activo para todo lo que llame el bloque y devuelve lo que devolvió el bloque. Un bloque que lanza deja un span en error con la excepción y la relanza sin cambios.

// 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
}

Un bloque cancelado es distinto, y se registra de otra forma. Una corrutina que su llamador o un tiempo límite abandona cierra el span como cancelled y no como error, porque un abandono deliberado no es un fallo de la operación y no debe leerse como tal en la lista de trazas.

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
}

Cuando un bloque no es la forma adecuada, inicia y cierra el span tú mismo.

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

Contexto

El span activo es un elemento del contexto de la corrutina, así que sigue a la corrutina y no al hilo. Una función suspend varias capas más abajo abre un hijo del padre correcto sin que ningún llamador se lo pase, y dos ejecuciones en el mismo dispatcher nunca ven los spans de la otra. Léelo con mcpd.currentSpan(), y obtén su cabecera con 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)
        "..."
    }

Entre servicios, el contexto viaja en una cabecera traceparent de W3C, que lleva consigo la decisión de muestreo, así que una traza nunca se registra a medias. Una cabecera mal formada inicia una traza nueva en vez de hacer fallar la solicitud que la traía.

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

Modelo, uso y errores

mcpd convierte los recuentos de tokens en costo en la ingesta, así que la cifra que ves en la traza es la cifra que te facturaron. Registrar un error pone el estado del span en error en el mismo movimiento, y siempre hay un mensaje: «algo falló» no sirve para actuar.

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

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

Los demás son setInput, setOutput, setMetadata, setAttribute, setAttributes, setTags, setCorrelation y addLink, además de child para un hijo explícito y event para un punto en el tiempo. Todos se pueden llamar sin riesgo desde cualquier hilo y ninguno lanza. Un span que pertenece al árbol de otro sistema de trazas lo nombra con externalParentSpanId, que se lleva para correlacionar y nunca se usa como padre dentro del árbol propio de mcpd.

Payloads, sinks y spans ya terminados

capturePayloads = false mantiene input y output dentro del proceso: se descartan antes de que el registro llegue a la cola, así que nada sensible sale en un flush, y un span puede anularlo para sí mismo. El exportador entrega cada lote a un Sink: el transporte HTTP por defecto, un RecordingSink que guarda los lotes en memoria para una prueba, o un ConsoleSink que escribe una línea JSON por señal. Un sink que lanza es un lote fallido, informado una vez por intervalo y nunca lanzado hacia quien llama.

// Keep prompts and completions in the process: the span still carries its
// timing, status, usage, model, 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()

Ejecuciones duraderas

El mismo cliente sirve las ejecuciones duraderas del daemon. Workflow construye la definición y guarda los cuerpos de los pasos como lambdas suspend; un Worker la registra, hace long polling para obtener leases y atiende cada uno en una corrutina, y startRun, watchRun, publishEvent y el resto son extensiones del cliente que reflejan la superficie de ejecuciones una a una. Un cuerpo lanza RetryableException para que se reintente dentro del presupuesto del paso, y registra spans bajo su paso a través del StepContext. Cada span abierto a través del contexto suma su uso al del paso según termina, así que un cuerpo que registra sus llamadas al modelo nunca suma tokens a mano; setUsage anula lo que sumaron los spans. Un Worker acepta cuatro hooks opcionales junto a onError (onLease, onStepStart, onStepEnd, onConflict) y responde a stats(): tomas, cuerpos en vuelo, finalizaciones, fallos por clase, conflictos por motivo y leases que no se pueden atender.

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

Qué hace cada construcción, el protocolo de leases y la API de ejecuciones están en Workflows, Workers y Ejecuciones.

Puntuaciones y feedback

min y max son parámetros obligatorios. Un siete es excelente sobre diez y malo sobre cien, así que un número sin escala declarada no se puede comparar con nada. El feedback es el juicio de una persona y se mantiene aparte de una puntuación, porque los dos responden preguntas distintas y nunca se deben promediar juntos.

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

Los logs y las métricas se adjuntan al span activo, así que un pico en un gráfico abre la ejecución que lo causó.

Flush y apagado

Registrar añade a una cola acotada en memoria y vuelve; una corrutina worker agrupa, comprime y envía. flush() envía lo que está en la cola y espera. shutdown() detiene el worker, envía lo que queda y se puede llamar dos veces sin problema. La exportación final se ejecuta sin poder cancelarse, porque al apagado se suele llegar desde un finally durante una cancelación y el último lote es el que explica por qué el proceso se está yendo.

Mcpd es AutoCloseable, así que funcionan tanto use como un shutdown hook. Una función serverless o un worker de vida corta que no llame a ninguno de los dos pierde su último lote, porque el runtime deja de planificar en cuanto el handler devuelve.

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

Lo que recibes de vuelta

Los métodos de lectura reflejan la API HTTP una a una y devuelven los tipos generados. Son los únicos métodos que lanzan, y lanzan McpdException: una consulta que estás esperando no es algo que se deba tragar.

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

Con qué puedes contar

• La instrumentación nunca lanza. Una exportación fallida, un servidor inalcanzable o un valor que no se serializa se informan como mucho una vez cada 30 segundos por java.util.logging y por lo demás se tragan. La observabilidad nunca debe tumbar aquello que observa.
• Registrar no bloquea, así que la latencia de tu servicio nunca depende de que mcpd sea rápido, ni de que esté en pie.
• La memoria está acotada. La cola guarda maxQueue señales; a partir de ahí se descartan las más antiguas y se cuentan, legibles como droppedSignals y en stats(). Un búfer sin límite convierte una caída de red en una muerte por falta de memoria.
• La decisión de muestreo se toma una vez, en la raíz, y se hereda.
• Un span se exporta cuando empieza y otra vez cuando termina, así que una ejecución que se cayó a mitad de vuelo sigue siendo visible. Cuando un span termina antes de que se haya enviado su registro de inicio, el registro se actualiza en el sitio y solo se envía uno.
• Los lotes se comprimen con gzip por encima de 1 KiB y llevan una clave de idempotencia que sobrevive a un reintento, así que una solicitud que expiró después de que el servidor la guardara no puede contar dos veces.
• Las duraciones se miden con un reloj monótono, así que un salto de NTP no puede producir una negativa.

Encontrar una ejecución cubre los filtros por completo. El modelo de datos lista todos los campos que llevan estos tipos.