Rust

Instalação

O SDK exporta em uma task tokio de segundo plano, então o que ele espera encontrar é um runtime. Ele roda sem um, e aí nada é enviado até você chamar flush(), que é o comportamento certo para uma ferramenta síncrona, mas não é o que um serviço de vida longa quer.

Construir o cliente

Um cliente por processo. Mcpd::new() lê o ambiente e fala com um mcpd local sem nada configurado. A construção nunca falha: uma configuração que o mcpd não consegue usar desliga o cliente e loga o motivo, porque uma chamada de observabilidade não é uma coisa de que um programa deva ter de tratar o erro.

// Nothing here is required. `Mcpd::new()` reads MCPD_ENDPOINT, MCPD_API_KEY
// and the rest from the environment and talks to a local mcpd by default.
// Anything set in code wins over the environment.
let mcpd = Mcpd::with_config(
    Config::from_env()
        .with_project("support")
        .with_environment("local")
        .with_service("support-agent", Some("1.0.0")),
);

Tudo que for definido no builder depois de Config::from_env() vence o ambiente.

Um ajuste que não faz o parse é ignorado com um aviso e o padrão é usado, porque um erro de digitação no ambiente de uma implantação não pode parar o programa. enabled = false transforma toda chamada em uma operação vazia que ainda devolve objetos válidos; os ids continuam bem formados e o traceparent continua se propagando, então um teste pode desligar o mcpd sem tocar no código que está sendo testado.

Registrar uma execução

Span é um guarda RAII. Ele termina quando é descartado, que é a forma com escopo em Rust: um bloco, um escopo, ou o fim da função. Chamar end() você mesmo faz a mesma coisa, e o guarda existe para que um retorno antecipado ou um panic não consiga deixar um span aberto e inacabado.

let run = mcpd.trace("support agent");
{
    let step = run.child_of("retrieve", SpanKind::Retrieval);
    step.set_input("order 5512");
    // Whatever this step does. The span ends at the closing brace, and it ends
    // on an early return, on a ? and on a panic as well.
}
run.end();

// The root span is the run. The sampling decision is made here, once, and
// every child inherits it.
let run = mcpd.trace("support agent");
run.set_input(question);
run.set_session("session-4f21");
run.set_user("customer-91");
run.set_tags(["support", "orders"]);

// A tool call. `run.child_of` nests it under the run.
let lookup = run.child_of("lookup_order", SpanKind::ToolCall);
lookup.set_input(json!({ "order_id": "5512" }));
let order = json!({ "status": "shipped", "carrier": "DHL", "eta": "Tuesday" });
lookup.set_output(&order);
lookup.end();

// A tool that fails. Recording the error sets the span's status to `error`,
// and a message is required: "something failed" is not actionable.
let notify = run.child_of("notify_customer", SpanKind::ToolCall);
notify.set_input(json!({ "channel": "sms" }));
notify.set_error("sms gateway timed out after 5s");
notify.end();
// The failure did not stop the run: the agent falls back to email, and the
// trace shows both the attempt and the recovery.
run.event_with("fell back to email", json!({ "after": "sms gateway timeout" }));

Contexto

Dentro de um processo, o contexto viaja em um task-local do tokio. Segurar um span não o torna o atual: Span::scope define o task-local para uma future e Span::sync_scope faz isso para uma closure. Dentro de qualquer um dos dois, Mcpd::span se liga à execução em andamento, então uma função auxiliar várias camadas abaixo aninha corretamente sem receber o pai. Span::child_of é a alternativa explícita para quando você já está com o pai em mãos.

// Opening a span does not make it current. scope() does.
let run = mcpd.trace("support agent");
run.scope(async {
    lookup(&mcpd).await;
})
.await;
run.end();

async fn lookup(mcpd: &Mcpd) {
    // No parent was threaded into this function, and this still nests under
    // the run, because the task-local carries it.
    let step = mcpd.span("lookup_order", SpanKind::ToolCall);
    step.set_output(serde_json::json!({ "status": "shipped" }));
}

Entre processos, o contexto viaja em um cabeçalho traceparent do W3C. A flag de amostragem viaja junto, então as duas metades de um trace concordam sobre o trace estar sendo guardado ou não.

// Outbound: put the header on the request.
let header = span.traceparent();

// Inbound: continue the caller's trace instead of starting a new one. An
// absent or malformed header starts a fresh trace rather than failing.
let run = mcpd.trace_from("charge card", inbound_header, SpanKind::AgentRun);

Modelo, uso e erros

As contagens de tokens são o que o mcpd precifica na ingestão, então o custo no trace é o custo que foi cobrado de você. Registrar um erro coloca o status do span em error no mesmo movimento, então os dois nunca podem discordar, e a mensagem é um argumento obrigatório, não uma opção.

generation.set_model("claude-sonnet-4", "anthropic");
generation.set_usage(TokenUsage {
    input_tokens: Some(412),
    output_tokens: Some(96),
    ..TokenUsage::default()
});
// Or, when the breakdown does not matter:
generation.set_tokens(412, 96);

// A message is required. "Something failed" is not actionable.
tool.set_error("sms gateway timed out after 5s");

// Or record the error you are already holding, which reads its message and kind.
tool.record_error(&err);

Os outros setters são set_input, set_output, set_attributes, set_metadata, set_tags, set_entity, set_session, set_user, set_status e set_error_detail. Todos recebem &self, então um span pode ser preenchido de qualquer lugar em que esteja emprestado.

Payloads, sinks e spans já concluídos

Config::with_capture_payloads(false) mantém prompts e respostas dentro do processo: os spans continuam carregando o seu tempo, o seu status, o seu uso, o seu modelo, a sua correlação e os seus erros, e input e output são descartados antes de chegar à fila. Um span sobrescreve o cliente para si mesmo com span.capture_payloads(..), nos dois sentidos.

O exportador entrega todo lote ao sink do cliente. O transporte HTTP é o padrão; Config::with_sink aceita um RecordingSink, que guarda os lotes em memória para um teste verificar o que foi registrado sem nenhum servidor, ou um ConsoleSink, que imprime uma linha JSON por sinal. Um sink que falha é uma exportação falha, reportada uma vez por intervalo e nunca levantada como exceção. mcpd.stats() responde queued, dropped, capacity, exported, failed e rejected.

mcpd.record(span) enfileira um span que já terminou, com ids, pai, início e fim incluídos, do jeito que um adaptador ou uma importação o tem: nunca reaninhado, nunca com os tempos refeitos nem reamostrado. Um registro malformado é descartado e contado como rejected. span.set_external_parent carrega um pai de um sistema de tracing que não é o mcpd, só para correlação.

Execuções duráveis

O mesmo cliente serve as execuções duráveis do daemon. Workflow::new constrói a definição e guarda os corpos das etapas como closures assíncronas; um Worker a registra, faz long polling por leases e serve cada um em uma task, e start_run, watch_run, publish_event e os demais espelham a superfície de execuções uma a uma. Um corpo devolve StepFailure::retryable para ser repetido dentro do orçamento da etapa, e registra spans abaixo da sua etapa através do StepContext. Todo span aberto assim, e todo span aninhado abaixo de um deles, soma o seu uso ao da etapa quando termina, então o daemon precifica o que o trace mostra sem o corpo somar tokens na mão; set_usage é a sobrescrita explícita. Os hooks on_lease, on_step_start, on_step_end e on_conflict do builder, junto com worker.stats(), tornam o worker observável por dentro.

let support = Workflow::new("support_ticket")
    .step(
        "classify",
        |ctx: StepContext| async move {
            let input: Value = ctx.input()?;
            let ticket = input["ticket"].as_str().unwrap_or("");
            let kind = if ticket.starts_with('Q') { "question" } else { "bug" };
            Ok::<_, StepFailure>(json!({ "kind": kind }))
        },
        StepOptions::new(),
    )
    .agent(
        "draft",
        |ctx: StepContext| async move {
            // Your own model call, recorded under this step in the run's trace.
            // Its usage is the step's usage: the daemon prices what the span spent.
            let span = ctx.span("model_call", SpanKind::ModelGeneration);
            span.set_model("claude-sonnet-4", "anthropic");
            span.set_tokens(812, 37);
            span.end();
            Ok::<_, StepFailure>("an answer")
        },
        StepOptions::new().input(draft_input),
    )
    .wait_for(
        "approval",
        "approved",
        "${input.ticket}",
        Duration::from_secs(7 * 24 * 3600),
        StepOptions::new().when(when("steps.classify.output.kind", "eq", "bug")),
    )
    .step(
        "send",
        |ctx: StepContext| async move {
            let input: Value = ctx.input()?;
            let text = input["text"].as_str().unwrap_or("");
            deliver(text, ctx.idempotency_key()).await.map_err(busy)?;
            Ok::<_, StepFailure>(json!({ "sent": true }))
        },
        StepOptions::new()
            .input(json!({ "text": "${steps.draft.output}" }))
            .at_most_once()
            .retry(backoff),
    )
    .output(json!({ "sent": "${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

Uma Scale é um argumento obrigatório, não uma opção. Um 0,9 solto, sem intervalo declarado, não pode ser comparado com nada, e quando alguém estiver lendo o gráfico já é tarde para perguntar o que ele queria dizer. Scale::UNIT é a de sempre, de 0.0 a 1.0. O feedback é o julgamento de uma pessoa, mantido à parte de uma nota porque os dois respondem perguntas diferentes e nunca devem entrar na mesma média.

// A score is a machine judgement. The scale is not optional: a bare 0.9 with
// no declared range cannot be compared to anything.
mcpd.score(
    Target::trace(&trace_id),
    "answer_relevance",
    0.9,
    Scale::UNIT,
    Some("Names the carrier and the delivery day the customer asked about."),
);

// Feedback is a person's judgement, kept separate from a score because the
// two answer different questions and must never be averaged together.
mcpd.feedback(Target::trace(&trace_id), FeedbackKind::Thumbs, "up", Some("fast and correct"));

Flush e encerramento

Registrar acrescenta a uma fila em memória e retorna. flush() envia o que está na fila; shutdown() faz o flush, para o worker e pode ser chamado duas vezes sem problema. Os dois podem ser aguardados, e os dois devolvem a última falha de exportação para você poder logá-la.

Um processo de vida curta precisa chamar um dos dois. Um binário que sai sem encerrar sai com os seus spans ainda na memória, e em uma função serverless o runtime congela o processo no instante em que o handler retorna, então a task de segundo plano nunca ganha outra vez.

#[tokio::main]
async fn main() {
    let mcpd = Mcpd::new();

    // ... the run ...

    // Ignoring this is a reasonable default: an observability failure is not
    // an application failure. Logging it costs one line.
    if let Err(e) = mcpd.shutdown().await {
        eprintln!("mcpd could not export: {e}");
    }
}

Os sinais perdidos em um flush que falhou não voltam para a fila. O transporte já tentou de novo, e uma fila que se enche outra vez a cada falha é como uma queda vira uma morte por falta de memória.

Com o que você pode contar

Com o daemon parado, o exemplo roda até o fim, imprime o que não conseguiu fazer, e sai com zero.

$ cargo run -p mcpd-sdk --example quickstart    Finished `dev` profile [unoptimized + debuginfo]     Running `target/debug/examples/quickstart`mcpd could not export: export failed: mcpd could not be reachedcould not read the trace back: mcpd could not be reached$ echo $?0

• Nenhuma chamada de instrumentação devolve erro ou entra em panic. Um mcpd inacessível, uma falha de serialização ou uma fila cheia são logados através do tracing no máximo uma vez por intervalo de flush e engolidos. A observabilidade nunca pode derrubar aquilo que ela observa.
• Registrar nunca bloqueia. A exportação acontece em uma task tokio de segundo plano, então a latência do seu agente não depende de o mcpd ser rápido, nem de estar no ar.
• A fila tem teto em max_queue. Quando ela enche, os sinais mais antigos são descartados e mcpd.stats().dropped conta cada um. Descartar os mais antigos é uma falha menor do que uma morte por falta de memória.
• Amostragem abaixo de 1.0 é determinística sobre o id do trace, em vez de aleatória, então todo processo de uma frota toma a mesma decisão sobre o mesmo trace e um trace que atravessa uma fronteira de serviço nunca fica registrado pela metade.
• Um span é exportado quando começa e atualizado quando termina, então uma execução que quebra no meio do caminho continua legível.

Ler de volta

O mesmo cliente lê. Cada método espelha um endpoint HTTP e devolve os tipos gerados, que vêm da especificação passando pelo mcpd-core e são reexportados como mcpd_sdk::model. As leituras devolvem um Result, porque uma consulta que você aguardou e que não pode ser respondida é notícia.

// The same client reads. This is the trace you just wrote.
match mcpd.get_trace(&trace_id, Include::all()).await {
    Ok(detail) => {
        println!("{}: {} spans", detail.trace.name, detail.spans.len());
        for span in &detail.spans {
            println!(
                "  {:<22} {:>8} {}",
                span.name,
                span.duration_ms().map_or("open".to_owned(), |ms| format!("{ms:.1}ms")),
                span.status,
            );
        }
        if let Some(cost) = detail.trace.cost.as_ref().and_then(|c| c.total_cost) {
            println!("  cost: {cost:.6} USD");
        }
        for score in detail.scores.iter().flatten() {
            println!("  score {}: {}", score.scorer_id, score.value);
        }
    }
    Err(e) => eprintln!("could not read the trace back: {e}"),
}

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