Kotlin

Instalação

Compilado com Kotlin 2.4, para JVM 17 ou posterior. As únicas dependências são kotlinx coroutines e serialization. O HTTP é o cliente do próprio JDK, então nada é acrescentado a um serviço que já tem uma opinião sobre HTTP.

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

Construir o cliente

Uma instância por processo. Todo campo de McpdConfig lê uma variável de ambiente como padrão e pode ser definido em código, então o mesmo código roda em desenvolvimento e em produção sem nenhuma edição. Mcpd() sem argumento nenhum funciona contra um 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",
    )
)

O código vence o ambiente, e o ambiente vence o padrão.

enabled = false transforma toda chamada de registro em uma operação vazia que ainda devolve spans válidos, então dá para desligar o mcpd nos testes sem tocar no código que está sendo testado. As leituras continuam indo até o servidor: uma consulta é algo que quem chamou está esperando, e uma resposta vazia seria uma mentira.

Registrar uma execução

span é a forma idiomática. Ela encerra o span, o torna o span ativo para tudo que o bloco chamar, e devolve o que o bloco devolveu. Um bloco que lança deixa um span error carregando a exceção e a relança sem alteração.

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

Um bloco cancelado é outra coisa, e é registrado de outro jeito. Uma corrotina abandonada por quem a chamou ou por um tempo limite encerra o span como cancelled, e não como error, porque um abandono deliberado não é uma falha da operação e não pode se ler como uma na lista de 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
}

Quando um bloco não é o formato certo, comece e encerre o span você mesmo.

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

O span ativo é um elemento do contexto da corrotina, então ele acompanha a corrotina, e não a thread. Uma função suspend várias camadas abaixo abre um filho do pai certo sem que nenhum chamador precise passar um, e duas execuções no mesmo dispatcher nunca enxergam os spans uma da outra. Leia-o com mcpd.currentSpan(), e pegue o seu cabeçalho com 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 serviços, o contexto viaja em um cabeçalho traceparent do W3C, que leva a decisão de amostragem junto, então um trace nunca fica registrado pela metade. Um cabeçalho malformado começa um trace novo em vez de falhar a requisição que o carregou.

// 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 e erros

As contagens de tokens são o que o mcpd transforma em custo na ingestão, então o número no trace é o número que foi cobrado de você. Registrar um erro coloca o status do span em error no mesmo movimento, e uma mensagem está sempre presente: "alguma coisa falhou" não dá para agir em cima.

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

Os demais são setInput, setOutput, setMetadata, setAttribute, setAttributes, setTags, setCorrelation e addLink, mais o child para um filho explícito e o event para um ponto no tempo. Todos podem ser chamados de qualquer thread com segurança e nenhum deles lança. Um span que pertence à árvore de outro tracer o nomeia com externalParentSpanId, carregado para correlação e nunca usado como pai dentro da árvore do próprio mcpd.

Payloads, sinks e spans já concluídos

capturePayloads = false mantém input e output dentro do processo: eles são descartados antes de o registro chegar à fila, então nada sensível sai em um flush, e um span sozinho pode sobrescrever isso para si mesmo. O exportador entrega todo lote a um Sink: o transporte HTTP por padrão, um RecordingSink que guarda os lotes em memória para um teste, ou um ConsoleSink que escreve uma linha JSON por sinal. Um sink que lança é um lote falho, reportado uma vez por intervalo e nunca levantado para quem chamou.

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

Execuções duráveis

O mesmo cliente serve as execuções duráveis do daemon. Workflow constrói a definição e guarda os corpos das etapas como lambdas suspensas; um Worker a registra, faz long polling por leases e serve cada um em uma corrotina, e startRun, watchRun, publishEvent e os demais são extensões do cliente que espelham a superfície de execuções uma a uma. Um corpo lança RetryableException para ser repetido dentro do orçamento da etapa, e registra spans abaixo da sua etapa através do StepContext. Todo span aberto pelo contexto soma o seu uso ao da etapa conforme termina, então um corpo que registra as suas chamadas de modelo nunca soma tokens na mão; setUsage sobrescreve o que os spans somaram. Um Worker aceita quatro hooks opcionais além de onError (onLease, onStepStart, onStepEnd, onConflict) e responde stats(): tomadas, corpos em andamento, conclusões, falhas por classe, conflitos por motivo e leases que não puderam ser servidos.

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

O que cada construção faz, o protocolo de lease e a API de execuções estão em Workflows, Workers e Execuções.

Notas e feedback

min e max são parâmetros obrigatórios. Sete é excelente em dez e ruim em cem, então um número sem escala declarada não pode ser comparado com nada. O feedback é o julgamento de uma pessoa e fica separado de uma nota, porque os dois respondem perguntas diferentes e nunca devem entrar na mesma média.

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

Os logs e as métricas se ligam ao span ativo, então um pico em um gráfico abre a execução que o causou.

Flush e encerramento

Registrar acrescenta a uma fila em memória com teto e retorna; uma corrotina worker agrupa em lotes, comprime e envia. flush() envia o que está na fila e espera. shutdown() para o worker, envia o que sobrou, e pode ser chamado duas vezes sem problema. A exportação final roda sem poder ser cancelada, porque o encerramento costuma ser alcançado a partir de um finally durante um cancelamento e o último lote é justamente o que explica por que o processo está indo embora.

Mcpd é AutoCloseable, então tanto o use quanto um shutdown hook funcionam. Uma função serverless ou um worker de vida curta que não chama nenhum dos dois perde o seu último lote, porque o runtime para de agendar no instante em que o handler retorna.

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

O que você recebe de volta

Os métodos de leitura espelham a API HTTP um a um e devolvem os tipos gerados. São os únicos métodos que lançam, e eles lançam McpdException: uma consulta pela qual você está esperando não é coisa para engolir.

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

Com o que você pode contar

• A instrumentação nunca lança. Uma exportação que falha, um servidor inacessível ou um valor que não serializa são reportados no máximo uma vez a cada 30 segundos através do java.util.logging e, fora isso, engolidos. A observabilidade nunca pode derrubar aquilo que ela observa.
• Registrar não bloqueia, então a latência do seu serviço nunca depende de o mcpd ser rápido, nem de estar no ar.
• A memória é limitada. A fila guarda maxQueue sinais; passando disso, os mais antigos são descartados e contados, legíveis em droppedSignals e em stats(). Um buffer sem limite transforma uma queda de rede em uma morte por falta de memória.
• A decisão de amostragem é tomada uma vez, na raiz, e herdada.
• Um span é exportado quando começa e de novo quando termina, então uma execução que quebrou no meio do caminho continua visível. Quando um span termina antes de o seu registro de início ter sido enviado, o registro é atualizado no lugar e só um é enviado.
• Os lotes são compactados com gzip acima de 1 KiB e carregam uma chave de idempotência que sobrevive a uma repetição, então uma requisição que estourou o tempo depois de o servidor já ter guardado o dado não pode contar duas vezes.
• As durações são medidas por um relógio monotônico, então um ajuste de NTP não consegue produzir uma duração negativa.

Encontrar uma execução cobre os filtros por completo. O modelo de dados lista cada campo que esses tipos carregam.