Rust

Install

The SDK exports on a background tokio task, so a runtime is what it expects to find. It runs without one, and then nothing is sent until you call flush(), which is the right behaviour for a synchronous tool but not what a long-lived service wants.

Construct the client

One client per process. Mcpd::new() reads the environment and talks to a local mcpd with nothing set at all. Construction never fails: a configuration mcpd cannot use disables the client and logs why, because an observability call is not a thing a program should have to handle an error from.

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

Anything set on the builder after Config::from_env() wins over the environment.

A setting that will not parse is skipped with a warning and the default is used, because a typo in a deployment's environment must not stop the program. enabled = false makes every call a no-op that still returns valid objects; ids are still well formed and traceparent still propagates, so a test can turn mcpd off without touching the code under test.

Record a run

Span is an RAII guard. It ends when it is dropped, which is the scoped form in Rust: a block, a scope, or the end of the function. Calling end() yourself does the same thing, and the guard is there so that an early return or a panic cannot leave a span open and unfinished.

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

Context

Inside a process, context travels in a tokio task-local. Holding a span does not make it current: Span::scope sets the task-local for a future and Span::sync_scope does it for a closure. Inside either, Mcpd::span attaches to the run in progress, so a helper several layers down nests correctly without being handed the parent. Span::child_of is the explicit alternative when you are already holding the parent.

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

Between processes, context travels in a W3C traceparent header. The sampling flag rides along with it, so the two halves of a trace agree about whether the trace is being kept.

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

Model, usage and errors

Token counts are what mcpd prices at ingest, so the cost on the trace is the cost you were billed. Recording an error sets the span's status to error in the same move, so the two can never disagree, and the message is a required argument rather than an option.

set_model_parameters records the settings the call was made with: temperature, top_p, top_k, max_tokens, seed, stop_sequences and has_response_format. They are settings rather than payload, so they survive with_capture_payloads(false), which is the case that matters: they are what somebody reproducing the run has to match. None is absent and Some(0) is a value, so a zero seed is a seed. The schema behind has_response_format is not recorded, only whether one was in force.

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

// How the model was called. Settings, not payload: kept with capture off.
generation.set_model_parameters(ModelParameters {
    temperature: Some(0.0),
    seed: Some(0),
    max_tokens: Some(1024),
    ..ModelParameters::default()
});

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

The rest of the setters are set_input, set_output, set_attributes, set_metadata, set_tags, set_entity, set_session, set_user, set_status and set_error_detail. All of them take &self, so a span can be filled in from anywhere it is borrowed.

Payloads, sinks and finished spans

Config::with_capture_payloads(false) keeps prompts and completions in the process: spans still carry their timing, status, usage, model, model parameters, correlation and errors, and input and output are dropped before they reach the queue. A span overrides the client for itself with span.capture_payloads(..), in either direction.

The exporter hands every batch to the client's sink. The HTTP transport is the default; Config::with_sink takes a RecordingSink, which keeps batches in memory so a test can assert what was recorded without a server, or a ConsoleSink, which prints one JSON line per signal. A sink that fails is a failed export, reported once per interval and never raised. mcpd.stats() answers queued, dropped, capacity, exported, failed and rejected.

mcpd.record(span) queues a span that already ended, ids, parent, start and end included, as an adapter or an import holds it: never re-parented, re-timed or re-sampled. A malformed record is dropped and counted as rejected. span.set_external_parent carries a parent from a tracing system that is not mcpd, for correlation only.

Durable runs

The same client serves the daemon's durable runs. Workflow::new builds the definition and keeps the step bodies as async closures; a Worker registers it, long-polls for leases and serves each one on a task, and start_run, watch_run, publish_event and the rest mirror the run surface one for one. A body returns StepFailure::retryable to be retried within the step's budget, and records spans under its step through the StepContext. Every span opened that way, and every span nested under one, adds its usage to the step when it ends, so the daemon prices what the trace shows without the body summing tokens by hand; set_usage is the explicit override. The builder's on_lease, on_step_start, on_step_end and on_conflict hooks and worker.stats() make the worker observable from inside.

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

What each construct does, the lease protocol and the run API are on Workflows, Workers and Runs.

Scores and feedback

A Scale is a required argument, not an option. A bare 0.9 with no declared range cannot be compared to anything, and by the time someone is reading the chart it is too late to ask what it meant. Scale::UNIT is the usual one, 0.0 to 1.0. Feedback is a person's judgement, kept apart from a score because the two answer different questions and must never be averaged together.

// 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 and shutdown

Recording appends to an in-memory queue and returns. flush() sends what is queued; shutdown() flushes, stops the worker and is safe to call twice. Both are awaitable, and both return the last export failure so you can log it.

A short-lived process must call one of them. A binary that exits without shutting down exits with its spans still in memory, and in a serverless function the runtime freezes the process the moment the handler returns, so the background task never gets another turn.

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

Signals lost to a failed flush are not re-queued. The transport has already retried, and a queue that refills itself on failure is how an outage becomes an out-of-memory kill.

What you can rely on

With the daemon stopped, the example runs to completion, prints what it could not do, and exits 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

• No instrumentation call returns an error or panics. An unreachable mcpd, a serialisation failure or a full queue is logged through tracing at most once per flush interval and swallowed. Observability must never take down the thing it observes.
• Recording never blocks. Export happens on a background tokio task, so your agent's latency does not depend on mcpd being fast, or up.
• The queue is capped at max_queue. When it is full the oldest signals are discarded and mcpd.stats().dropped counts them. Dropping the oldest is a smaller failure than an out-of-memory kill.
• Sampling below 1.0 is deterministic on the trace id rather than random, so every process in a fleet makes the same decision about the same trace and a trace that crosses a service boundary is never half recorded.
• A span is exported when it starts and updated when it ends, so a run that crashes mid-flight is still readable.

Reading it back

The same client reads. Every method mirrors one HTTP endpoint and returns the generated types, which come from the specification by way of mcpd-core and are re-exported as mcpd_sdk::model. Reads return a Result, because a query you awaited and cannot answer is news.

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

Finding a run covers the filters in full. The data model lists every field these types carry.