Rust

Instalación

El SDK exporta en una tarea de tokio en segundo plano, así que espera encontrar un runtime. Funciona sin él, y entonces no se envía nada hasta que llamas a flush(), que es el comportamiento correcto para una herramienta síncrona pero no lo que quiere un servicio de larga vida.

Construir el cliente

Un cliente por proceso. Mcpd::new() lee el entorno y habla con un mcpd local sin que haya nada configurado. La construcción nunca falla: una configuración que mcpd no puede usar desactiva el cliente y deja anotado por qué, porque una llamada de observabilidad no es algo de lo que un programa deba tener que gestionar un error.

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

Todo lo que se asigne en el builder después de Config::from_env() gana al entorno.

Un ajuste que no se puede parsear se omite con un aviso y se usa el valor por defecto, porque una errata en el entorno de un despliegue no debe detener el programa. enabled = false convierte cada llamada en una operación vacía que sigue devolviendo objetos válidos; los ids siguen estando bien formados y traceparent se sigue propagando, así que una prueba puede apagar mcpd sin tocar el código bajo prueba.

Registrar una ejecución

Span es un guard RAII. Termina cuando se suelta, que es la forma con ámbito en Rust: un bloque, un ámbito o el final de la función. Llamar tú mismo a end() hace lo mismo, y el guard está para que un retorno temprano o un panic no puedan dejar un span abierto y sin terminar.

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 un proceso, el contexto viaja en un task-local de tokio. Tener un span no lo convierte en el actual: Span::scope asigna el task-local para un future y Span::sync_scope lo hace para un closure. Dentro de cualquiera de los dos, Mcpd::span se engancha a la ejecución en curso, así que una función auxiliar varias capas más abajo se anida bien sin recibir el padre. Span::child_of es la alternativa explícita cuando ya tienes el padre en la mano.

// 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 procesos, el contexto viaja en una cabecera traceparent de W3C. La marca de muestreo va con ella, así que las dos mitades de una traza coinciden sobre si la traza se está guardando.

// 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 y errores

mcpd calcula el precio a partir de los recuentos de tokens en la ingesta, así que el costo que ves en la traza es el costo que te facturaron. Registrar un error pone el estado del span en error en el mismo movimiento, así que los dos nunca pueden discrepar, y el mensaje es un argumento obligatorio y no una opción.

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

Los demás setters son set_input, set_output, set_attributes, set_metadata, set_tags, set_entity, set_session, set_user, set_status y set_error_detail. Todos reciben &self, así que un span se puede rellenar desde cualquier sitio donde esté prestado.

Payloads, sinks y spans ya terminados

Config::with_capture_payloads(false) mantiene los prompts y las completions dentro del proceso: los spans siguen llevando sus tiempos, su estado, su uso, su modelo, su correlación y sus errores, y input y output se descartan antes de llegar a la cola. Un span anula al cliente para sí mismo con span.capture_payloads(..), en cualquiera de los dos sentidos.

El exportador entrega cada lote al sink del cliente. El transporte HTTP es el valor por defecto; Config::with_sink acepta un RecordingSink, que guarda los lotes en memoria para que una prueba pueda comprobar lo que se registró sin servidor, o un ConsoleSink, que imprime una línea JSON por señal. Un sink que falla es una exportación fallida, informada una vez por intervalo y nunca lanzada. mcpd.stats() responde queued, dropped, capacity, exported, failed y rejected.

mcpd.record(span) encola un span que ya terminó, con sus ids, su padre, su inicio y su fin, tal como los tiene un adaptador o una importación: nunca se le cambia el padre, ni los tiempos, ni el muestreo. Un registro mal formado se descarta y se cuenta como rejected. span.set_external_parent lleva un padre de un sistema de trazas que no es mcpd, solo para correlacionar.

Ejecuciones duraderas

El mismo cliente sirve las ejecuciones duraderas del daemon. Workflow::new construye la definición y guarda los cuerpos de los pasos como closures async; un Worker la registra, hace long polling para obtener leases y atiende cada uno en una tarea, y start_run, watch_run, publish_event y el resto reflejan la superficie de ejecuciones una a una. Un cuerpo devuelve StepFailure::retryable para que se reintente dentro del presupuesto del paso, y registra spans bajo su paso a través del StepContext. Cada span abierto así, y cada span anidado bajo uno de ellos, suma su uso al del paso cuando termina, de modo que el daemon cobra lo que muestra la traza sin que el cuerpo sume tokens a mano; set_usage es la anulación explícita. Los hooks on_lease, on_step_start, on_step_end y on_conflict del builder y worker.stats() hacen que el worker se pueda observar desde 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}" }));

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

Puntuaciones y feedback

Una Scale es un argumento obligatorio, no una opción. Un 0.9 suelto sin rango declarado no se puede comparar con nada, y para cuando alguien esté leyendo el gráfico ya es tarde para preguntar qué significaba. Scale::UNIT es la habitual, de 0.0 a 1.0. 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 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 y apagado

Registrar añade a una cola en memoria y vuelve. flush() envía lo que está en la cola; shutdown() hace flush, detiene el worker y se puede llamar dos veces sin problema. Los dos se pueden esperar, y los dos devuelven el último fallo de exportación para que lo puedas anotar.

Un proceso de vida corta tiene que llamar a uno de los dos. Un binario que sale sin apagar el cliente sale con sus spans todavía en memoria, y en una función serverless el runtime congela el proceso en cuanto el handler devuelve, así que la tarea en segundo plano no vuelve a tener turno.

#[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}");
    }
}

Las señales que se pierden en un flush fallido no vuelven a la cola. El transporte ya ha reintentado, y una cola que se rellena sola cuando falla es la forma en que una caída se convierte en una muerte por falta de memoria.

Con qué puedes contar

Con el daemon detenido, el ejemplo llega hasta el final, imprime lo que no pudo hacer y sale con cero.

$ 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

• Ninguna llamada de instrumentación devuelve un error ni hace panic. Un mcpd inalcanzable, un fallo de serialización o una cola llena se anotan por tracing como mucho una vez por intervalo de flush y se tragan. La observabilidad nunca debe tumbar aquello que observa.
• Registrar nunca bloquea. La exportación ocurre en una tarea de tokio en segundo plano, así que la latencia de tu agente no depende de que mcpd sea rápido, ni de que esté en pie.
• La cola está limitada a max_queue. Cuando está llena se descartan las señales más antiguas y mcpd.stats().dropped las cuenta. Descartar las más antiguas es un fallo menor que una muerte por falta de memoria.
• Un muestreo por debajo de 1.0 es determinista sobre el id de traza en vez de aleatorio, así que todos los procesos de una flota toman la misma decisión sobre la misma traza y una traza que cruza un límite de servicio nunca se registra a medias.
• Un span se exporta cuando empieza y se actualiza cuando termina, así que una ejecución que se cae a mitad de vuelo sigue siendo legible.

Volver a leerlo

El mismo cliente lee. Cada método refleja un endpoint HTTP y devuelve los tipos generados, que salen de la especificación pasando por mcpd-core y se reexportan como mcpd_sdk::model. Las lecturas devuelven un Result, porque una consulta que esperaste y no se puede responder es una noticia.

// 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 una ejecución cubre los filtros por completo. El modelo de datos lista todos los campos que llevan estos tipos.