Rust

Installation

Le SDK exporte sur une tâche tokio de fond : c’est donc un runtime qu’il s’attend à trouver. Il fonctionne sans, et rien n’est alors envoyé tant que vous n’appelez pas flush(), ce qui est le bon comportement pour un outil synchrone mais pas ce qu’attend un service de longue durée.

Construire le client

Un client par processus. Mcpd::new() lit l’environnement et s’adresse à un mcpd local sans que rien ne soit défini. La construction n’échoue jamais : une configuration que mcpd ne peut pas utiliser désactive le client et journalise pourquoi, car un appel d’observabilité n’est pas quelque chose dont un programme devrait avoir à traiter l’erreur.

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

Tout ce qui est défini sur le builder après Config::from_env() l’emporte sur l’environnement.

Un réglage qui refuse de s’analyser est ignoré avec un avertissement et la valeur par défaut est retenue, car une faute de frappe dans l’environnement d’un déploiement ne doit pas arrêter le programme. enabled = false fait de chaque appel une opération sans effet qui renvoie quand même des objets valides ; les identifiants restent bien formés et traceparent se propage toujours, si bien qu’un test peut couper mcpd sans toucher au code testé.

Enregistrer une exécution

Span est un garde RAII. Il se termine quand il est droppé, ce qui est la forme à portée délimitée en Rust : un bloc, une portée, ou la fin de la fonction. Appeler end() vous-même fait la même chose, et le garde est là pour qu’un retour anticipé ou une panique ne puisse pas laisser un span ouvert et inachevé.

let run = mcpd.trace("support agent");
{
    let step = run.child_of("retrieve", SpanKind::Retrieval);
    step.set_input("order 5512");
    // Peu importe ce que fait cette étape. Le span se termine à l’accolade
    // fermante, et aussi sur un retour anticipé, sur un ? et sur une panique.
}
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" }));

Contexte

À l’intérieur d’un processus, le contexte voyage dans un task-local tokio. Détenir un span ne le rend pas courant : Span::scope pose le task-local pour un futur et Span::sync_scope le fait pour une closure. À l’intérieur de l’un ou de l’autre, Mcpd::span se rattache à l’exécution en cours, si bien qu’une fonction utilitaire située plusieurs couches plus bas s’imbrique correctement sans qu’on lui ait remis le parent. Span::child_of est l’alternative explicite quand vous détenez déjà le parent.

// Ouvrir un span ne le rend pas courant. scope() le fait.
let run = mcpd.trace("support agent");
run.scope(async {
    lookup(&mcpd).await;
})
.await;
run.end();

async fn lookup(mcpd: &Mcpd) {
    // Aucun parent n’a été passé à cette fonction, et cela s’imbrique quand
    // même sous l’exécution, car le task-local le transporte.
    let step = mcpd.span("lookup_order", SpanKind::ToolCall);
    step.set_output(serde_json::json!({ "status": "shipped" }));
}

Entre processus, le contexte voyage dans un en-tête W3C traceparent. Le drapeau d’échantillonnage l’accompagne, si bien que les deux moitiés d’une trace s’accordent sur le fait qu’elle est conservée ou non.

// En sortie, posez l’en-tête sur la requête.
let header = span.traceparent();

// En entrée, continuez la trace de l’appelant au lieu d’en démarrer une
// nouvelle. Un en-tête absent ou mal formé démarre une nouvelle trace plutôt
// que d’échouer.
let run = mcpd.trace_from("charge card", inbound_header, SpanKind::AgentRun);

Modèle, consommation et erreurs

mcpd tarife les décomptes de tokens à l’ingestion : le coût porté par la trace est donc celui qui vous a été facturé. Enregistrer une erreur place du même geste le statut du span sur error, si bien que les deux ne peuvent jamais se contredire, et le message est un argument obligatoire plutôt qu’une option.

generation.set_model("claude-sonnet-4", "anthropic");
generation.set_usage(TokenUsage {
    input_tokens: Some(412),
    output_tokens: Some(96),
    ..TokenUsage::default()
});
// Ou, quand le détail n’a pas d’importance.
generation.set_tokens(412, 96);

// Un message est obligatoire. « Quelque chose a échoué » ne dit pas quoi faire.
tool.set_error("sms gateway timed out after 5s");

// Ou enregistrez l’erreur déjà en main, dont le message et le type sont lus.
tool.record_error(&err);

Les autres setters sont set_input, set_output, set_attributes, set_metadata, set_tags, set_entity, set_session, set_user, set_status et set_error_detail. Tous prennent &self : un span peut donc être renseigné depuis partout où il est emprunté.

Payloads, sinks et spans terminés

Config::with_capture_payloads(false) garde les prompts et les complétions dans le processus : les spans portent toujours leurs temps, leur statut, leur consommation, leur modèle, leur corrélation et leurs erreurs, et input et output sont écartés avant d’atteindre la file d’attente. Un span change le réglage du client pour lui-même avec span.capture_payloads(..), dans un sens comme dans l’autre.

L’exporter remet chaque lot au sink du client. Le transport HTTP est le sink par défaut ; Config::with_sink accepte un RecordingSink, qui garde les lots en mémoire pour qu’un test vérifie ce qui a été enregistré sans serveur, ou un ConsoleSink, qui imprime une ligne JSON par signal. Un sink qui échoue est un export en échec, signalé une fois par intervalle et jamais levé. mcpd.stats() répond sur queued, dropped, capacity, exported, failed et rejected.

mcpd.record(span) met en file un span déjà terminé, avec les identifiants, le parent, le début et la fin, tel que le détient un adaptateur ou un import : jamais re-parenté, re-daté ni ré-échantillonné. Un enregistrement mal formé est écarté et compté comme rejected. span.set_external_parent porte un parent venu d’un système de tracing autre que mcpd, pour la seule corrélation.

Exécutions durables

Le même client dessert les exécutions durables du daemon. Workflow::new construit la définition et retient le corps des étapes sous forme de closures asynchrones ; un Worker l’enregistre, attend des leases en long polling et sert chacun d’eux sur une tâche, et start_run, watch_run, publish_event et les autres reprennent un pour un la surface des exécutions. Un corps d’étape renvoie StepFailure::retryable pour être réessayé dans le budget de l’étape, et enregistre des spans sous son étape via le StepContext. Chaque span ouvert de cette manière, et chaque span imbriqué sous l’un d’eux, ajoute sa consommation à l’étape quand il se termine : le daemon tarife donc ce que montre la trace sans que le corps totalise les tokens à la main ; set_usage est la substitution explicite. Les hooks on_lease, on_step_start, on_step_end et on_conflict du builder, ainsi que worker.stats(), rendent le worker observable de l’intérieur.

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

Ce que fait chaque construction, le protocole de lease et l’API des exécutions sont décrits dans Workflows, Workers et Exécutions.

Scores et feedback

Une Scale est un argument obligatoire, pas une option. Un 0.9 tout nu, sans intervalle déclaré, ne peut être comparé à rien, et quand quelqu’un lit le graphique il est trop tard pour demander ce qu’il voulait dire. Scale::UNIT est la plus courante, de 0.0 à 1.0. Le feedback est le jugement d’une personne, tenu à l’écart d’un score parce que les deux répondent à des questions différentes et ne doivent jamais être moyennés ensemble.

// 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 et arrêt

Enregistrer ajoute à une file d’attente en mémoire et rend la main. flush() envoie ce qui est en file ; shutdown() vide la file, arrête le worker et peut être appelé deux fois sans risque. Les deux s’attendent avec await, et les deux renvoient le dernier échec d’export pour que vous puissiez le journaliser.

Un processus de courte durée doit appeler l’un des deux. Un binaire qui sort sans s’arrêter proprement sort avec ses spans encore en mémoire, et dans une fonction serverless le runtime gèle le processus dès que le handler rend la main : la tâche de fond n’a donc jamais son prochain tour.

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

    // ... l’exécution ...

    // L’ignorer est un choix par défaut raisonnable, car un échec
    // d’observabilité n’est pas un échec de l’application. Le journaliser coûte
    // une ligne.
    if let Err(e) = mcpd.shutdown().await {
        eprintln!("mcpd could not export: {e}");
    }
}

Les signaux perdus lors d’un flush en échec ne sont pas remis en file. Le transport a déjà réessayé, et une file qui se remplit à nouveau après un échec est la façon dont une panne devient un arrêt pour mémoire épuisée.

Ce sur quoi vous pouvez compter

Avec le daemon arrêté, l’exemple va jusqu’au bout, affiche ce qu’il n’a pas pu faire, et sort avec le code zéro.

$ 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

• Aucun appel d’instrumentation ne renvoie d’erreur ni ne panique. Un mcpd injoignable, un échec de sérialisation ou une file pleine sont journalisés via tracing au plus une fois par intervalle de flush, puis avalés. L’observabilité ne doit jamais faire tomber ce qu’elle observe.
• L’enregistrement ne bloque jamais. L’export a lieu sur une tâche tokio de fond : la latence de votre agent ne dépend donc pas de la vitesse de mcpd, ni de sa disponibilité.
• La file est plafonnée à max_queue. Quand elle est pleine, les signaux les plus anciens sont jetés et mcpd.stats().dropped les compte. Jeter les plus anciens est un échec moindre qu’un arrêt pour mémoire épuisée.
• Un échantillonnage inférieur à 1.0 est déterministe sur l’identifiant de trace plutôt qu’aléatoire : tous les processus d’une flotte prennent donc la même décision sur la même trace, et une trace qui franchit une frontière de service n’est jamais enregistrée à moitié.
• Un span est exporté quand il démarre et mis à jour quand il se termine, si bien qu’une exécution qui plante en plein vol reste lisible.

Relire ce qui a été enregistré

Le même client lit. Chaque méthode reflète un endpoint HTTP et renvoie les types générés, qui viennent de la spécification par l’intermédiaire de mcpd-core et sont réexportés sous mcpd_sdk::model. Les lectures renvoient un Result, car une requête que vous avez attendue et à laquelle il est impossible de répondre est une nouvelle qui compte.

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

Retrouver une exécution couvre les filtres en détail. Le modèle de données liste chaque champ que ces types portent.