智能体走的每一步,都保存到运行结束,并且留下记录。
mcpd 是守护进程,不是框架。步骤之间的一切由它保存:重试、定时器、等待审批的那一整周;每个步骤内部的工作由你自己的代码完成,语言不限。每一步都会进入一条 trace,并带上它的成本。
构建并运行
git clone https://github.com/mcpd/mcpd && cd mcpd && make build && ./target/release/mcpd demo 然后打开 localhost:4317。无需账号,不用另外搭数据库,也不用签任何东西。目前还没有发布版本:镜像和软件包会随第一个版本一起发布。阅读五分钟快速开始。
一次读得懂的运行。
智能体出错的地方,不是写着报错的那行日志。它在三步之前就错了:模型选错了工具,然后又选了一次。
mcpd 把整次运行还原成一棵树:每次模型调用、每次工具调用、每次检索,各自附上当时确切的提示词和补全、各自的延迟,以及各自的成本。
- agent support_agent fail
- model chat
- tool returns_policy
- model chat
- tool returns_policy
- model chat
- tool returns_policy
The model called one tool three times and never answered. That is visible in four seconds here, and invisible in a log file.
运行记录鼓
同一次运行,绕在记录鼓上。时间沿圆周展开,每个 span 一条轨道。转动它,笔尖读出下方的 span。
一次运行,比启动它的进程活得更久。
步骤之间,这次运行归守护进程持有。步骤内部的一切归你的进程所有,什么也不用交出去:模型不用,提示词不用,凭据也不用。
租约板
守护进程在板上为每个步骤保留一个标签。worker 只有在心跳持续到达时才持有标签。停止该 worker,经过一个租约期和一个回收窗口后,标签回到板上,交给另一个 worker。
图是数据,步骤主体归你
工作流用代码写一次。图作为一份定义交给守护进程,由它校验并记录版本。步骤主体留在你的进程里,六种语言任选,交给一个 worker 来执行。
重试、定时器和等待都是数据行
临时性失败会按该步骤自己的预算重试。睡眠或等待审批时不需要任何进程活着,这次运行照样保住,跨过一次发布、一次崩溃,或者一个星期。事件按名称和键回应等待,重复的事件算重复投递,不会变成第二次运行。
每次运行都是一条 trace
每个步骤都是一个 span,带着它的输入、输出和上报的用量,用给其他 span 定价的同一份价目表计价。步骤内部的模型调用挂在它下面,所以搜索、成本和评分器面对一次运行时不用改动。
用你已经在写的语言接入。
六个 SDK,一套行为。记录一个 span 不会阻塞、不会抛异常,也不会无限增长:mcpd 不可达时你的智能体察觉不到,而这是可观测性唯一可以接受的失败方式。每个 SDK 还自带一个 worker,所以持久化运行的各个步骤,由记录它们的同一个进程来执行。
它们都不是必需的。把 OTEL_EXPORTER_OTLP_ENDPOINT 指向 mcpd,你已经在跑的 OpenTelemetry SDK 就能把数据送到,应用里不会有我们的一行代码。这六个 SDK 只是 HTTP 接口之上的便利封装,不是被记录的前提。
跟着框架一起交付的可观测性,只能记录用那个框架写的智能体。mcpd 接收任何到达它 OTLP 端点的数据:一个智能体框架、你自己写的循环,或者两者调用的服务。
接入一次运行
/**
* One ticket, one trace.
*
* `withSpan` opens the span, keeps it active for everything the callback
* starts, and ends it when the callback settles. Nothing inside has to be
* passed a handle to be nested under it.
*/
async function handleTicket(ticket: Ticket): Promise<{ answer: string; traceId: string }> {
let traceId = ''
const answer = await mcpd.withSpan('support_agent', async (run) => {
// Keep the id: a score or a piece of feedback that arrives later attaches
// to this run by it.
traceId = run.traceId
// Tags label the whole run, so the trace list can filter on them.
run.setTags(['support', 'tier-1'])
run.setCorrelation({ sessionId: ticket.id, userId: ticket.customerId })
run.setInput({ question: ticket.question })
const order = await lookUpOrder(ticket)
const answer = await draftAnswer(ticket, order)
run.setOutput({ answer })
return answer
})
return { answer, traceId }
} # The root span is the run. Everything opened inside it becomes a child,
# because the active span lives in a `contextvars.ContextVar`.
with mcpd.trace("support_agent", session_id="sess-8121", user_id=user_id,
tags=["support", "orders"]) as run:
run.set_input({"question": question})
# A model call. `set_usage` is what mcpd prices the run from, so the
# cost on the trace is the cost you were actually billed.
with mcpd.span("plan", kind=SpanKind.MODEL_GENERATION) as generation:
generation.set_model("gpt-4o-mini", provider="openai")
generation.set_input([
{"role": "system", "content": "You are a support agent."},
{"role": "user", "content": question},
])
time.sleep(0.12) # stand-in for the provider round trip
generation.set_output("I should look the order up first.")
generation.set_usage(input_tokens=812, output_tokens=96, cache_read=128)
# A tool call that fails. The `with` block records the exception on the
# span, marks it as an error, and lets the exception through unchanged.
order = None
try:
with mcpd.span("lookup_order", kind=SpanKind.TOOL_CALL) as tool:
tool.set_entity("tool", entity_id="orders.lookup", name="Order lookup")
tool.set_input({"order_id": "A-7741"})
order = lookup_order("A-7741")
tool.set_output(order)
except TimeoutError as exc:
# The span already carries the error. This is the agent deciding
# what to do about it, which is a different question.
mcpd.log("warn", "order lookup failed, answering from context", {"error": str(exc)})
mcpd.metric("support.tool_failures", 1, "counter", labels={"tool": "orders.lookup"})
reply = ("I could not reach the order system just now. Your order A-7741 "
"shipped on Tuesday and tracking should update within the hour.")
run.set_output(reply)
run.set_metadata({"used_fallback": order is None})
return reply, run.trace_id // 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" })); // handleTicket is the agent. It drafts an answer with a model, tries to send
// it with a tool, and fails on the send so the error path is recorded too.
func handleTicket(ctx context.Context, m *mcpd.Client, question string) string {
ctx, run := m.Start(ctx, "support_agent",
mcpd.WithSpanKind(mcpd.SpanKindAgentRun),
mcpd.WithEntity(mcpd.EntityKindAgent, "support-agent", "Support agent"),
mcpd.WithSession("sess-4821"),
mcpd.WithUser("u-317"),
mcpd.WithTags("support", "billing"),
mcpd.WithInput(question),
)
defer run.End()
m.Log(ctx, mcpd.LogLevelInfo, "ticket received", map[string]any{"channel": "email"})
answer := draftAnswer(ctx, m, question)
run.SetOutput(answer)
if err := sendReply(ctx, m, answer); err != nil {
// The span is now an error with a message, which is what makes it
// worth reading. End below records the failed status.
run.RecordError(err, mcpd.ErrorDomain("email"), mcpd.ErrorRetryable(true))
}
return run.TraceID()
} val question = "where is my order 41?"
// `span` starts the operation, ends it, and records anything it throws.
// Inside the block the span is the active one, so the helpers below find
// their parent through the coroutine context rather than being handed it.
val traceId = mcpd.span(
"support_agent",
SpanKind.AGENT_RUN,
input = question,
sessionId = "session-8f21",
userId = "customer-4477",
tags = listOf("support", "email"),
) { run ->
mcpd.log(LogLevel.INFO, "handling a question", mapOf("channel" to "email"))
val plan = planAnswer(mcpd, question)
val order = lookUpOrder(mcpd, 41)
val estimate = estimateDelivery(mcpd, 41)
val answer = buildString {
append(plan).append(' ').append(order)
if (estimate != null) append(" It should arrive ").append(estimate).append('.')
}
run.setOutput(answer)
mcpd.metric("support.answers", 1.0, MetricKind.COUNTER, unit = "{answer}")
// The block's value is the value of `span`.
run.traceId
} question = "where is my order 41?"
# The block's value is the value of `span`. Inside it, this span is the active
# one for the process, so the helpers below find their parent without being
# handed it.
trace_id =
Mcpd.span mcpd, "support_agent",
kind: :agent_run,
input: question,
session_id: "session-8f21",
user_id: "customer-4477",
tags: ["support", "email"] do
Mcpd.log(mcpd, :info, "handling a question", %{"channel" => "email"})
plan = plan_answer.(mcpd, question)
order = look_up_order.(mcpd, 41)
estimate = estimate_delivery.(mcpd, 41)
_answer =
[plan, order, estimate && "It should arrive #{estimate}."]
|> Enum.reject(&is_nil/1)
|> Enum.join(" ")
Mcpd.metric(mcpd, "support.answers", 1, kind: :counter, unit: "{answer}")
Mcpd.Context.current().trace_id
end # One batch carries any mix of the five signals. Send the whole run at once.
curl -s -X POST "$MCPD/v1/ingest" \
-H 'content-type: application/json' \
-H "authorization: Bearer ${MCPD_API_KEY:-}" \
-d @- <<JSON
{
"ingest_key": "run-$TRACE",
"spans": [
{
"trace_id": "$TRACE",
"span_id": "$ROOT",
"name": "support_agent",
"kind": "agent_run",
"status": "ok",
"start_time": "$NOW", "end_time": "$NOW",
"is_root": true, "is_event": false, "received_at": "$NOW",
"input": { "question": "Where is order 4417?" },
"output": { "answer": "It is in the Leeds depot." },
"correlation": { "project": "default", "environment": "production" }
},
{
"trace_id": "$TRACE",
"span_id": "$GEN",
"parent_span_id": "$ROOT",
"name": "chat",
"kind": "model_generation",
"status": "ok",
"start_time": "$NOW", "end_time": "$NOW",
"is_root": false, "is_event": false, "received_at": "$NOW",
"model": "gpt-4o",
"provider": "openai",
"usage": { "input_tokens": 1284, "output_tokens": 210 },
"correlation": { "project": "default" }
}
]
}
JSON
# → {"accepted":2,"rejected":0,"duplicate":0}
#
# mcpd prices the model span from its own price book on the way in, so cost is
# stored with the run rather than recomputed later from rates that may have
# changed since. 六种信号,由产生它们的那次运行关联起来。
多数工具把 trace 放在一处,把日志放在另一处,剩下的时间戳留给你用眼睛对。mcpd 把六种信号都存在同一组 trace id 和 span id 上,所以图表上的一个尖峰能打开引发它的那次运行,一次运行也能列出它输出的日志行。
分拣台
六种信号混在同一条传送带上到达。读取头读出每件上的 trace id,同一次运行的每一件都落进这次运行的抽屉。
13 种语言,说法一致。
Cross-language conformance
tests/e2e/run.mjs
| Channel | root | tree | status | error | model | tokens | score | feedback | ids |
|---|---|---|---|---|---|---|---|---|---|
| TypeScript | |||||||||
| Python | |||||||||
| Go | |||||||||
| Elixir | |||||||||
| Kotlin | |||||||||
| Rust | |||||||||
| cURL | |||||||||
| TypeScript | |||||||||
| Python | |||||||||
| Go | |||||||||
| Elixir | |||||||||
| Kotlin | |||||||||
| Rust |
13 languages, one canonical run, one daemon. Every trace comes back the same shape.
481 checks
每个 SDK 都单独测过,那只能证明它做到了作者的本意,说明不了几个 SDK 之间是否一致。所以让它们各自向同一个守护进程写入同一次运行,再把 trace 读回来相互比对。
某个行为在一种语言里成立、在另一种语言里不成立,只有在这里才会暴露出来。每次改动都会在 CI 里跑一遍。与现有方案的对比。
谁读了什么,包括谁被挡了回去。
决定一次安全评审的问题,不是这些数据安不安全,而是事后你能不能拿出证据,说明谁访问过它。
Audit log
GET /v1/api/audit
| Time | Principal | Action | Project | Result |
|---|---|---|---|---|
| 14:02:11 | a3f9c2e1b7d4 | get_trace | checkout | ok |
| 14:02:11 | a3f9c2e1b7d4 | list_spans | checkout | ok |
| 14:04:38 | 7c1e08b53a9f | list_dataset_items | billing | refused |
| 14:07:02 | d52a4f8c0e63 | capture_into_dataset | checkout | ok |
| 14:11:55 | d52a4f8c0e63 | delete_trace | checkout | ok |
| 14:19:20 | 7c1e08b53a9f | list_feedback | billing | ok |
A refusal is recorded like a read. The principal is a digest of the key, never the key. Retention never trims this table.
18 actions
六种密钥权限
发送遥测数据、读取提示词、修改已存数据、销毁数据、启动运行和领取步骤,是不同的权限。不设置 write 时它接受 ingest 密钥,不设置 admin 时它接受 read 密钥。不设置 execute 和 worker 时,它们接受 write 密钥。升级不会把你挡在门外。
项目就是边界
密钥、保留期和采样都按项目设置。列在某个项目下的密钥只能访问那个项目,守护进程会先确认一份数据归谁所有,然后才返回其中任何一条。
不会向外回传
整个守护进程里只有四处会发出站请求,每一处都由你配置:转发器、两个 webhook 和裁判模型。没有更新检查,没有遥测,也没有许可证服务器。
哪把密钥,哪个人
审计日志会写明是哪把密钥。把你的入口网关已经设置的请求头,以及允许设置它的地址交给 mcpd,它还会写明是哪个人。从其他地方发来的请求头,只是任何人都能提出的说法,所以那一栏宁可留空,也不写一个猜测。
一次删除请求就是一次调用
一次 DELETE 会删除记在某个人名下的每一次运行,连同挂在它们下面的 span、日志、指标、评分和反馈,并按信号分别给出数量。该密钥打不开的项目里的数据,既不会被改动,也不会被计入。
先检查,再信任
mcpd doctor 会给一套部署评级:存储能不能写进去再读回来,是否有权限被留在开放状态,配置过的项目是否收到过 span。任何需要你处理的问题,它都会指出修好它的那个参数。
mcpd 不做什么。
每个工具都有这样一份清单。多数把它留在内部,等你用到第二周才发现。这份清单就放在这里,给的是原因,不是日期。
- 不支持 gRPC 上的 OTLP。
- HTTP 用两种编码都能传输 trace,日志和指标用 JSON。gRPC 完全不支持。
- 没有按厂商命名的导出器。
- mcpd 转发的是 OTLP,一套实现就能到达多数目的地,也让 mcpd 只是一跳,而不是你的数据必须留下的地方。只讲自家协议的目的地则到不了。
- 没有 SSO 或 SCIM。
- 密钥的作用域是项目,那是租户边界,不是身份提供方。一个人的身份,取决于 mcpd 前面的入口网关认定了什么。
- 没有多区域摄取。
- 在 Postgres 或 ClickHouse 上,它以一个服务后面的若干副本运行。那不是跨区域故障转移。
完整的说明在 PARITY.md 里,对照的是对现有产品的一次直接源码审计。它在功能落地时更新,而不是在功能列入计划时更新。
五分钟拿到第一条 trace。
跑起守护进程,把智能体指向它,打开界面。如果用的时间比这更长,那就是一个缺陷,我们希望收到报告。
开始
docker run -p 4317:4317 ghcr.io/mcpd/mcpd