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() 之后由构建器设置的任何东西,都优先于环境变量。
| 字段 | 环境变量 | 默认值 |
|---|---|---|
endpoint | MCPD_ENDPOINT | http://localhost:4317 |
api_key | MCPD_API_KEY | 无 |
project | MCPD_PROJECT | default |
environment | MCPD_ENVIRONMENT | 无 |
service_name | MCPD_SERVICE_NAME | 无 |
service_version | MCPD_SERVICE_VERSION | 无 |
sample | MCPD_SAMPLE | 1.0 |
batch_size | MCPD_BATCH_SIZE | 128 |
flush_interval | MCPD_FLUSH_INTERVAL_MS | 2s |
max_queue | MCPD_MAX_QUEUE | 8192 |
timeout | MCPD_TIMEOUT_MS | 10s |
enabled | MCPD_ENABLED | true |
capture_payloads | MCPD_CAPTURE_PAYLOADS | true |
log_level | MCPD_LOG_LEVEL | debug |
解析不了的设置会被跳过,同时记一条警告,然后用默认值,因为一次部署的环境
变量里打错一个字,不该把程序停下来。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}"),
}