Rust

Installazione

L’SDK esporta su un task tokio in secondo piano, quindi si aspetta di trovare un runtime. Gira anche senza, e allora non parte niente finché non chiami flush(), che è il comportamento giusto per uno strumento sincrono ma non quello che vuole un servizio di lunga durata.

Costruire il client

Un client per processo. Mcpd::new() legge l’ambiente e parla con un mcpd locale senza che tu imposti niente. La costruzione non fallisce mai: una configurazione che mcpd non sa usare disattiva il client e scrive in un log il perché, perché una chiamata di osservabilità non è una cosa da cui un programma debba gestire un errore.

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

Tutto quello che imposti sul builder dopo Config::from_env() vince sull’ambiente.

Un’impostazione che non si lascia interpretare viene saltata con un avvertimento e si usa il valore predefinito, perché un errore di battitura nell’ambiente di un deploy non deve fermare il programma. enabled = false rende ogni chiamata un’operazione a vuoto che restituisce comunque oggetti validi; gli identificatori restano ben formati e il traceparent si propaga lo stesso, così un test può spegnere mcpd senza toccare il codice sotto test.

Registrare un’esecuzione

Span è una guardia RAII. Finisce quando viene rilasciata, che è la forma con ambito in Rust: un blocco, uno scope, o la fine della funzione. Chiamare end() per conto tuo fa la stessa cosa, e la guardia sta lì perché un ritorno anticipato o un panic non possano lasciare uno span aperto e mai concluso.

let run = mcpd.trace("support agent");
{
    let step = run.child_of("retrieve", SpanKind::Retrieval);
    step.set_input("order 5512");
    // Qualunque cosa faccia questo passo. Lo span finisce alla parentesi che
    // chiude, e finisce anche su un ritorno anticipato, su un ? e su un panic.
}
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" }));

Contesto

Dentro un processo il contesto viaggia in un task-local di tokio. Tenere in mano uno span non lo rende quello corrente: Span::scope imposta il task-local per un future e Span::sync_scope lo fa per una closure. Dentro l’uno o l’altra, Mcpd::span si attacca all’esecuzione in corso, così una funzione di appoggio parecchi livelli più in basso si annida correttamente senza che le venga passato il padre. Span::child_of è l’alternativa esplicita per quando il padre lo hai già in mano.

// Aprire uno span non lo rende quello corrente. Lo fa scope().
let run = mcpd.trace("support agent");
run.scope(async {
    lookup(&mcpd).await;
})
.await;
run.end();

async fn lookup(mcpd: &Mcpd) {
    // In questa funzione non è stato passato nessun padre, e lo span si annida
    // comunque sotto l'esecuzione, perché il task-local se lo porta dietro.
    let step = mcpd.span("lookup_order", SpanKind::ToolCall);
    step.set_output(serde_json::json!({ "status": "shipped" }));
}

Fra due processi il contesto viaggia in un header traceparent W3C. Il flag di campionamento viaggia nello stesso header, così le due metà di una traccia vanno d’accordo sul fatto che la traccia venga tenuta o no.

// In uscita: metti l'header sulla richiesta.
let header = span.traceparent();

// In entrata: continua la traccia di chi chiama invece di aprirne una nuova.
// Un header assente o malformato apre una traccia nuova anziché fallire.
let run = mcpd.trace_from("charge card", inbound_header, SpanKind::AgentRun);

Modello, consumo ed errori

I conteggi dei token sono quelli che mcpd valorizza all’ingest, quindi il costo che vedi sulla traccia è il costo che ti è stato fatturato. Registrare un errore porta nello stesso gesto lo stato dello span a error, così i due non possono mai contraddirsi, e il messaggio è un argomento obbligatorio e non un’opzione.

generation.set_model("claude-sonnet-4", "anthropic");
generation.set_usage(TokenUsage {
    input_tokens: Some(412),
    output_tokens: Some(96),
    ..TokenUsage::default()
});
// Oppure, quando il dettaglio non conta:
generation.set_tokens(412, 96);

// Il messaggio è obbligatorio. "Qualcosa è fallito" non dice che cosa fare.
tool.set_error("sms gateway timed out after 5s");

// Oppure registra l'errore che hai già in mano: ne legge messaggio e tipo.
tool.record_error(&err);

Gli altri setter sono set_input, set_output, set_attributes, set_metadata, set_tags, set_entity, set_session, set_user, set_status e set_error_detail. Prendono tutti &self, quindi uno span si può riempire da qualunque punto in cui è preso in prestito.

Payload, sink e span già conclusi

Config::with_capture_payloads(false) tiene i prompt e le completion dentro il processo: gli span portano lo stesso i loro tempi, lo stato, il consumo, il modello, la correlazione e gli errori, e input e output vengono scartati prima di arrivare alla coda. Uno span sovrascrive il client per sé con span.capture_payloads(..), in tutte e due le direzioni.

L’exporter consegna ogni batch al sink del client. Il trasporto HTTP è quello predefinito; Config::with_sink prende un RecordingSink, che tiene i batch in memoria così un test può verificare che cosa è stato registrato senza un server, oppure un ConsoleSink, che stampa una riga JSON per segnale. Un sink che fallisce è un export fallito, riferito una volta per intervallo e mai sollevato come eccezione. mcpd.stats() risponde queued, dropped, capacity, exported, failed e rejected.

mcpd.record(span) mette in coda uno span già concluso, con identificatori, padre, inizio e fine compresi, così come li ha un adattatore o un’importazione: senza che gli si cambi mai il padre, i tempi o il campionamento. Un record malformato viene scartato e contato come rejected. span.set_external_parent porta un padre che viene da un sistema di tracing che non è mcpd, solo per la correlazione.

Esecuzioni durevoli

Lo stesso client serve le esecuzioni durevoli del daemon. Workflow::new costruisce la definizione e conserva i corpi dei passi come closure asincrone; un Worker la registra, fa long polling per ottenere i lease e serve ognuno di essi su un task, e start_run, watch_run, publish_event e gli altri rispecchiano uno a uno la superficie delle esecuzioni. Un corpo restituisce StepFailure::retryable per essere ritentato entro il budget del suo passo, e registra span sotto il proprio passo attraverso lo StepContext. Ogni span aperto così, e ogni span annidato sotto di esso, somma il proprio consumo a quello del passo quando finisce, così il daemon valorizza quello che la traccia mostra senza che il corpo sommi i token a mano; set_usage è l’override esplicito. Gli hook on_lease, on_step_start, on_step_end e on_conflict del builder e worker.stats() rendono il worker osservabile da 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}" }));

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

Punteggi e feedback

Una Scale è un argomento obbligatorio, non un’opzione. Un 0.9 nudo, senza un intervallo dichiarato, non si può confrontare con niente, e quando qualcuno sta leggendo il grafico è troppo tardi per chiedere che cosa volesse dire. Scale::UNIT è quella solita, da 0.0 a 1.0. Il feedback è il giudizio di una persona, tenuto da parte rispetto a un punteggio perché i due rispondono a domande diverse e non vanno mai mediati insieme.

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

Registrare aggiunge a una coda in memoria e ritorna. flush() manda quello che è in coda; shutdown() fa il flush, ferma il worker e si può chiamare due volte senza rischi. Tutti e due si possono attendere, e tutti e due restituiscono l’ultimo export fallito, così lo puoi scrivere in un log.

Un processo di breve durata deve chiamarne uno. Un binario che esce senza arrestare il client esce con i suoi span ancora in memoria, e in una funzione serverless il runtime congela il processo nel momento in cui l’handler ritorna, quindi il task in secondo piano non ha mai un altro turno.

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

    // ... l'esecuzione ...

    // Ignorarlo è una scelta ragionevole: un guasto di osservabilità non è un
    // guasto dell'applicazione. Scriverlo in un log costa una riga.
    if let Err(e) = mcpd.shutdown().await {
        eprintln!("mcpd could not export: {e}");
    }
}

I segnali persi per un flush fallito non tornano in coda. Il trasporto ha già riprovato, e una coda che si riempie di nuovo da sola a ogni guasto è il modo in cui un’interruzione diventa un processo ucciso per memoria esaurita.

Su che cosa puoi contare

Con il daemon spento, l’esempio arriva in fondo, stampa quello che non è riuscito a fare, ed esce con 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

• Nessuna chiamata di strumentazione restituisce un errore o va in panic. Un mcpd irraggiungibile, un guasto di serializzazione o una coda piena finiscono in un log attraverso tracing al massimo una volta per intervallo di flush e poi vengono inghiottiti. L’osservabilità non deve mai far cadere la cosa che osserva.
• Registrare non blocca mai. L’export avviene su un task tokio in secondo piano, quindi la latenza del tuo agente non dipende da quanto mcpd sia veloce, né dal fatto che sia attivo.
• La coda ha un tetto a max_queue. Quando è piena i segnali più vecchi vengono scartati e mcpd.stats().dropped li conta. Buttare i più vecchi è un guasto più piccolo di un processo ucciso per memoria esaurita.
• Il campionamento sotto 1.0 è deterministico sull’id di traccia anziché casuale, così ogni processo di una flotta prende la stessa decisione sulla stessa traccia e una traccia che attraversa il confine fra due servizi non è mai registrata a metà.
• Uno span viene esportato quando parte e aggiornato quando finisce, così un’esecuzione che va in crash per strada resta leggibile.

Rileggerla

Lo stesso client legge. Ogni metodo rispecchia un endpoint HTTP e restituisce i tipi generati, che vengono dalla specifica passando per mcpd-core e sono riesportati come mcpd_sdk::model. Le letture restituiscono un Result, perché una query che hai atteso e a cui non si sa rispondere è una notizia.

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

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