跳到主要内容
mcpd
无守护进程
源码

Rust

span 是一个守卫,被 drop 时就结束,所以一次提前返回或者一个错误,都不会让它一直开着。上下文走的是一个 tokio task-local。

安装

SDK 在一个后台 tokio 任务上做导出,所以它期望能找到一个运行时。没有运行时 它也能跑,只是在你调 flush() 之前什么都不会发出去;这对一个 同步的小工具是对的行为,但不是一个长期运行的服务想要的。

cargo add mcpd-sdk

构造客户端

一个进程一个客户端。Mcpd::new() 读环境变量,什么都不设置时就和 本地的 mcpd 通信。构造永远不会失败:mcpd 用不了的配置会把客户端禁用掉,并把 原因记进日志,因为一次可观测性调用,不该是一个程序还得去处理它的错误的东西。

连接

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

在 Config::from_env() 之后由构建器设置的任何东西,都优先于环境变量。

字段环境变量默认值
endpointMCPD_ENDPOINThttp://localhost:4317
api_keyMCPD_API_KEY无
projectMCPD_PROJECTdefault
environmentMCPD_ENVIRONMENT无
service_nameMCPD_SERVICE_NAME无
service_versionMCPD_SERVICE_VERSION无
sampleMCPD_SAMPLE1.0
batch_sizeMCPD_BATCH_SIZE128
flush_intervalMCPD_FLUSH_INTERVAL_MS2s
max_queueMCPD_MAX_QUEUE8192
timeoutMCPD_TIMEOUT_MS10s
enabledMCPD_ENABLEDtrue
capture_payloadsMCPD_CAPTURE_PAYLOADStrue
log_levelMCPD_LOG_LEVELdebug

解析不了的设置会被跳过,同时记一条警告,然后用默认值,因为一次部署的环境 变量里打错一个字,不该把程序停下来。enabled = false 让每次调用 都变成空操作,但仍然返回合法的对象;id 依然是良构的, traceparent 依然照常传播,所以一个测试不用碰被测代码,就能把 mcpd 关掉。

记录一次运行

Span 是一个 RAII 守卫。它在被 drop 时结束,这就是 Rust 里带 作用域的那种写法:一个代码块、一个作用域,或者函数的末尾。自己调 end() 做的是同一件事;守卫在那里,是为了让一次提前返回或者 一次 panic,不会留下一个开着、没结束的 span。

span 是一个守卫

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

上下文

在一个进程内部,上下文走的是一个 tokio task-local。拿着一个 span 并不会让它 成为当前的那个:Span::scope 为一个 future 设置这个 task-local, Span::sync_scope 为一个闭包做同样的事。在它们两个里面, Mcpd::span 会挂到正在进行的那次运行上,所以往下好几层的一个 辅助函数,不用谁把父级递给它,也能正确地嵌套。当你手上已经拿着父级时, Span::child_of 就是那个显式的写法。

那个 task-local

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

在进程之间,上下文走 W3C traceparent 请求头。采样标志跟着它一起 走,所以一条 trace 的两半,对这条 trace 是不是被留下这件事看法一致。

跨过服务边界

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

模型、用量与错误

mcpd 在摄取时就是按 token 数定价的,所以 trace 上的成本就是你被计费的那个 成本。记录一个错误的同时会把这个 span 的状态设成 error,所以 这两者不可能对不上;而那条消息是一个必填参数,不是一个可选项。

一个 span 带着什么

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

其余的设置方法是 set_input、set_output、 set_attributes、set_metadata、 set_tags、set_entity、set_session、 set_user、set_status 和 set_error_detail。它们都收 &self,所以一个 span 在任何借到它的地方都能往里填东西。

载荷、sink 与已经结束的 span

Config::with_capture_payloads(false) 把提示词和补全留在进程里: span 照样带着它的时间、状态、用量、模型、关联信息和错误,而 input 和 output 在进入队列之前就被丢掉。单个 span 可以用 span.capture_payloads(..) 为自己覆盖客户端的设置,两个 方向都可以。

导出器把每个批次交给客户端的 sink。默认是 HTTP 传输层; Config::with_sink 收下一个 RecordingSink,它把 批次留在内存里,于是一个测试不需要服务器就能断言记下了什么;或者收下一个 ConsoleSink,它把每个信号打印成一行 JSON。失败的 sink 算一次 失败的导出,每个间隔上报一次,绝不抛出。mcpd.stats() 会回答 queued、dropped、capacity、 exported、failed 和 rejected。

mcpd.record(span) 把一个已经结束的 span 排进队列,连同 id、 父级、开始和结束时间,按一个适配器或者一次导入手上的样子来:绝不重新指定 父级、重新计时或者重新采样。格式不对的记录会被丢弃,并计入 rejected。span.set_external_parent 携带的是一个 来自非 mcpd 的 tracing 系统的父级,只用于关联。

持久化运行

同一个客户端也服务守护进程的持久化运行。Workflow::new 构建定义, 并把那些步骤主体作为异步闭包持有;一个 Worker 注册它,用长轮询 要租约,并在一个任务上逐份服务,而 start_run、 watch_run、publish_event 以及其余的方法与运行接口 一一对应。步骤主体返回 StepFailure::retryable,表示要在这个 步骤的预算内重试,并通过 StepContext 把 span 记在自己这个步骤 下面。这样开出来的每一个 span,以及嵌套在其中任何一个下面的每一个 span, 结束时都会把自己的用量加到这个步骤上,所以守护进程按 trace 显示出来的东西 定价,步骤主体不用手工去加 token;set_usage 是那个显式的覆盖。 构建器上的 on_lease、on_step_start、 on_step_end 和 on_conflict 这几个钩子,加上 worker.stats(),让这个 worker 从内部也是可观测的。

一个工作流

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

每种构造做什么、租约协议和运行 API,在 工作流、 Worker 和 运行这几页上。

评分与反馈

Scale 是一个必填参数,不是一个可选项。一个没有声明范围的裸 0.9 没法和任何东西比较,而等到有人在看图的时候,再问它当初是什么意思已经 太晚了。常用的是 Scale::UNIT,也就是 0.0 到 1.0。反馈是人的 判断,和评分分开放,因为这两者回答的是不同的问题,绝不能被平均到一起。

评分与反馈

// 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() 把排着队的 东西发出去;shutdown() 先刷出,再停掉 worker,而且调两次也是 安全的。两者都可以 await,也都会返回最后一次导出失败,这样你可以把它记进 日志。

一个短命的进程必须调它们其中一个。一个没有关闭就退出的二进制,退出时 span 还留在内存里;而在一个 serverless 函数里,处理函数一返回,运行时就把进程 冻住,于是那个后台任务再也轮不上了。

关闭

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

因为一次刷出失败而丢掉的信号不会重新排队。传输层已经重试过了,而一个失败时 自己往回填的队列,正是一次故障变成一次内存耗尽被杀的路子。

你可以依赖的东西

在守护进程停着的情况下,这个示例照样跑完,把它做不到的事情打印出来,然后以 0 退出。

没有守护进程在跑的时候
$ 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
  • 没有哪个接入调用会返回错误,或者 panic。一个不可达的 mcpd、一次序列化 失败,或者一个满了的队列,都会通过 tracing 每个刷出间隔最多 记一条日志,然后被吞掉。可观测性绝不能把它观测的那个东西搞垮。
  • 记录从不阻塞。导出发生在一个后台 tokio 任务上,所以你的智能体的延迟,不 取决于 mcpd 快不快,或者活没活着。
  • 队列的上限是 max_queue。队列满时最旧的信号会被扔掉, mcpd.stats().dropped 记下它们的数量。丢掉最旧的,是一个比 内存耗尽被杀小得多的失败。
  • 采样率低于 1.0 时,采样是按 trace id 确定的,而不是随机的,所以一个集群 里的每个进程,对同一条 trace 都会做出同样的决定,一条跨过服务边界的 trace 也绝不会只记了一半。
  • 一个 span 在开始时就导出,在结束时更新,所以一次跑到半路崩掉的运行仍然 读得出来。

把它读回来

读也是同一个客户端。每个方法对应一个 HTTP 端点,返回生成出来的类型;这些 类型经由 mcpd-core 从规范来,并以 mcpd_sdk::model 重新导出。读取返回一个 Result,因为一次你等着结果、却答不上来 的查询是个消息。

检索与分析

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

查找一次运行把过滤条件 讲全了。数据模型 列出了这些类型带的每一个字段。