TypeScript
一个没有依赖的包。它记录 span、日志、指标、评分和反馈,而且从不从一次接入调用里抛出异常。
安装
Node 20 及以上、Deno、Bun、浏览器和边缘运行时。以 ESM 为主,另外带一份给
require 用的 CommonJS 构建。只有两处是 Node 专有的,
async_hooks 和 zlib,它们在运行时才加载,在没有
它们的地方跳过。
npm install @mcpd/sdk 构造客户端
一个进程一个客户端。每个选项也都会读一个环境变量,所以同一份代码不用改一个
字,就能在开发环境和生产环境里跑。完全不带参数时,它连的是
http://localhost:4317 上的守护进程。
连接
import { Mcpd } from '@mcpd/sdk'
// Every field here also reads an environment variable, so the same code runs
// unchanged in development and in production. With no arguments at all this
// talks to a daemon on http://localhost:4317.
const mcpd = new Mcpd({
project: 'support',
environment: 'development',
serviceName: 'support-agent',
serviceVersion: '1.4.0',
}) 代码优先于环境变量,环境变量优先于默认值。
| 选项 | 环境变量 | 默认值 |
|---|---|---|
endpoint | MCPD_ENDPOINT | http://localhost:4317 |
apiKey | MCPD_API_KEY | 无 |
project | MCPD_PROJECT | default |
environment | MCPD_ENVIRONMENT | 无 |
serviceName | MCPD_SERVICE_NAME | 无 |
serviceVersion | MCPD_SERVICE_VERSION | 无 |
sample | MCPD_SAMPLE | 1.0 |
enabled | MCPD_ENABLED | true |
capturePayloads | MCPD_CAPTURE_PAYLOADS | true |
logLevel | MCPD_LOG_LEVEL | debug |
batchSize | MCPD_BATCH_SIZE | 128 |
flushIntervalMs | MCPD_FLUSH_INTERVAL_MS | 2000 |
maxQueue | MCPD_MAX_QUEUE | 8192 |
timeoutMs | MCPD_TIMEOUT_MS | 10000 |
sink | HTTP 传输层 |
enabled: false 什么都不记录,也什么都不发送,而每次调用仍然返回
合法的对象,列表读取则返回空的一页。一个测试套件不用碰任何一行接入代码,就能
把 mcpd 关掉。
capturePayloads: false 把提示词和补全留在进程里。span 照样记录,
带着它的时间、状态、用量、模型、关联信息和错误;input 和
output 在进入队列之前就被丢掉,所以一次刷出不可能带走任何敏感
内容。单个 span 可以在自己的启动选项里用 capturePayloads 覆盖
这个设置,两个方向都可以。
记录一次运行
withSpan 是带作用域的那种写法。它开一个 span,让这个 span 对
回调启动的一切保持活动,并在回调落定时结束它。回调抛出异常时,这个 span 会
被标记为 error,带上抛出的那条消息,异常本身则原样再抛出去,
所以你自己的错误处理不会因为被观测而改变。
记录一次运行
/**
* 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 }
} 上下文
在 Node 和 Bun 上,当前活动的 span 存在一个 AsyncLocalStorage
里。在一个 withSpan 回调里启动的任何东西,都不需要拿到引用就会
嵌套在它下面;在同一个事件循环上交错运行的两次运行,也永远看不到对方的 span。
嵌套
// Anything the callback starts nests under it. No handle is passed down.
await mcpd.withSpan('support_agent', async (run) => {
const order = await mcpd.withSpan(
'lookup_order',
async (tool) => {
tool.setInput({ orderId: 'A-1042' })
mcpd.log('info', 'this line belongs to lookup_order')
return { status: 'in transit' }
},
{ kind: 'tool_call' },
)
run.setOutput(order)
})
浏览器和大多数边缘运行时没有 AsyncLocalStorage。在那里,SDK 改成
只保存一个值。它嵌套是对的,在一次运行内部也挺得过一次 await,
但它分不出两次并发的运行,结果会是一次运行的工具调用出现在另一次运行的树里。
在这些运行时上并发跑智能体时,请显式地把父级传下去。
显式指定父级
// Correct on every runtime, including ones with no AsyncLocalStorage.
const run = mcpd.trace('support_agent')
const step = run.child('call_model', { kind: 'model_generation' })
step.end()
await run.withChild('call_tool', async (tool) => {
tool.setOutput({ ok: true })
})
run.end()
// Which mechanism is in use, at runtime.
mcpd.stats().context // 'async-local-storage' | 'fallback'
跨服务时,上下文走 W3C traceparent 请求头。一条 trace 能挺过一次
网络跳转,靠的全部就是它。格式不对的头会被忽略,并开一条新的根 trace,因为
一个谁都读不出来的头,不构成让携带它的那个请求失败的理由。
跨过服务边界
// Outbound: put the active span on the request.
await fetch(url, {
headers: { traceparent: span.traceparent() },
})
// Inbound: continue that trace instead of starting a new one.
const run = mcpd.trace('handle_request', {
traceparent: request.headers.get('traceparent'),
}) 模型、用量与错误
mcpd 在摄取时就是按 token 数定价的,所以 trace 上的成本就是你被计费的那个
成本。请上报供应商告诉你的数字,而不是自己去数 token。错误一定带一条消息,
而设置错误的同时会把这个 span 的状态设成 error,所以这两者不
可能对不上。
一个 span 带着什么
generation.setModel('claude-sonnet-4', 'anthropic')
generation.setUsage({
inputTokens: 812,
outputTokens: 37,
inputDetails: { cacheRead: 640 },
})
// A failure you caught and handled yourself.
generation.setError('the provider timed out', {
kind: 'TimeoutError',
retryable: true,
})
// A point in time inside the span, with no duration of its own.
generation.event('first_token', { kind: 'model_chunk' }) 持久化运行
同一个客户端也服务守护进程的持久化运行。workflow() 构建定义并
持有那些步骤主体;一个 Worker 注册它,用长轮询要租约并逐份服务,
而 mcpd.runs、mcpd.events 和
mcpd.definitions 与运行接口一一对应。步骤主体抛出
RetryableError 表示要在这个步骤的预算内重试,并通过
StepContext 把 span 记在自己这个步骤下面。这样记下的 span,
以及嵌套在它下面的每一个 span,结束时都会把自己的用量加到这个步骤上,所以
守护进程按 trace 显示出来的东西定价,步骤主体从不用手工去加 token;
ctx.setUsage 是一次显式的覆盖。onError 旁边还有
四个钩子:onLease、onStepStart、
onStepEnd 和 onConflict,而
worker.stats() 统计领取数、完成数、按类别分的失败数,以及按原因
分的冲突数。
一个工作流
const support = workflow({ id: 'support_ticket' })
.step('classify', async (ctx, input: { ticket: string }) => ({
kind: input.ticket.startsWith('Q') ? 'question' : 'bug',
}))
.agent(
'draft',
async (ctx, input: { kind: string }) => {
// Your own model call, recorded under this step in the run's trace. Its
// usage adds up into the step's, which the daemon prices.
return await ctx.withSpan('model_call', async (span) => {
span.setModel('claude-sonnet-4', 'anthropic')
span.setUsage({ inputTokens: 812, outputTokens: 37 })
return `an answer to a ${input.kind}`
})
},
{ input: { kind: '${steps.classify.output.kind}' } },
)
.waitFor('approval', {
event: 'approved',
key: '${input.ticket}',
timeoutMs: 7 * 24 * 3_600_000,
when: when('steps.classify.output.kind', 'eq', 'bug'),
})
.step(
'send',
async (ctx, input: { text: string }) => {
if (!(await deliver(input.text, ctx.idempotencyKey))) {
throw new RetryableError('mail relay busy', { retryAfterMs: 2_000 })
}
return { sent: true }
},
{
input: { text: '${steps.draft.output}' },
effect: 'at_most_once',
retry: { maxAttempts: 3, backoff: 'exponential', initialMs: 500 },
},
)
.output({ sent: '${steps.send.output.sent}' }) 每种构造做什么、租约协议和运行 API,在 工作流、 Worker 和 运行这几页上。
评分与反馈
min 和 max 是类型强制要求的。一个没有声明量程的
0.7 分没法和任何东西比较,而等到有人在看图的时候,再问它当初是什么意思已经
太晚了。反馈是人的判断,和评分分开存放,因为这两者回答的是不同的问题,绝不
能被平均到一起。
评分与反馈
// A machine evaluation of the run. `min` and `max` are required: 0.86 on an
// undeclared scale cannot be compared to anything.
mcpd.score({
traceId,
scorerId: 'answer_relevance',
value: 0.86,
min: 0,
max: 1,
reason: 'Answered the question and gave the delivery date.',
})
// A human evaluation. Kept separate from a score, and never averaged with
// one, because the two answer different questions.
mcpd.feedback({
traceId,
kind: 'thumbs',
value: 'up',
comment: 'Exactly what the customer asked for.',
authorKind: 'end_user',
})
// Anything you already count can travel with the trace.
mcpd.metric('support.tickets_handled', 1, 'counter', { labels: { queue: 'tier-1' } })
不指定 trace 时,两者都会挂到当前活动 span 所在的那条 trace 上,所以你自己
产品里的一个赞成或反对按钮,只要拿着你留下的那个 trace id 就能调
feedback。
五种信号,一套 id
span、日志、指标、评分和反馈走的是同一个队列,带着同样的 trace id 和 span id。正是这一点,让图上的一个尖峰能打开引起它的那次运行。
-
Spans
What ran, nested, with timings
-
Logs
Lines the run emitted
-
Metrics
Numbers, with exemplars back to a run
-
Scores
Machine evaluation
-
Feedback
What a person thought
-
Aspect notes
A sentence a model wrote about the run
刷出与关闭
记录就是往队列里追加一条。在后台 worker 把它发出去之前,没有任何东西离开过
这个进程,而它每两秒发一次,或者在一个批次装满时发。flush() 和
shutdown() 都可以 await,也都是幂等的,shutdown()
还会把那个 worker 停掉。
在一个 serverless 函数里,这件事不是可选的。你一返回响应,运行时就把进程 冻住,后台 worker 再也不会跑,于是最后一个批次就丢了。请在响应发出去之前 await 一次刷出。
一个 serverless 处理函数
export default async function handler(request: Request): Promise<Response> {
const answer = await mcpd.withSpan('answer', async (run) => {
run.setInput(await request.text())
return 'ok'
})
// The batch is still in memory here. Send it before the response goes out.
await mcpd.flush()
return new Response(answer)
}
在 Node 上,客户端还会在 beforeExit 时刷出一次,所以一个忘了调
shutdown() 的脚本,最后那个批次仍然发得出去。这个钩子在显式的
process.exit() 上,或者在一个致命信号上,都不会触发,所以它是
一张安全网,不是替代品。
断言你记下了什么
RecordingSink 用内存替掉网络,于是一个测试可以把被测代码产生的
span、日志、指标、评分和反馈读回来,不需要服务器,也不需要端口。
ConsoleSink 则每个信号打印一行 JSON,而任何带
export(batch, ingestKey) 方法的对象都是一个 sink。抛异常的
sink 算一个失败的批次加一条诊断,绝不会变成你代码里的异常。
mcpd.stats() 会回答 queued、dropped、
capacity、exported、failed 和
rejected。
一个不需要服务器的测试
import { Mcpd, RecordingSink } from '@mcpd/sdk'
const sink = new RecordingSink()
const mcpd = new Mcpd({ sink })
await answerQuestion(mcpd)
await mcpd.flush()
expect(sink.spans.map((s) => s.name)).toEqual(['answer_question', 'model_call'])
expect(sink.scores[0]?.scorerId).toBe('answer_relevance') 记录一个已经结束的 span
mcpd.record(span) 把一个 span 排进队列,它的 id、父级和时间都是
你已经拿在手上的:也就是另一套 tracing 系统的适配器、一次批量导入或者一次
重放手上有的那些东西。它按给的样子排队,绝不会被重新指定父级、重新计时或者
重新采样,其中留空的关联信息则从客户端补上。id 不是 OpenTelemetry 那个形状、
名字为空,或者缺少开始时间的记录会被丢弃,并计入
stats().rejected。在任何一个 span 上,
externalParentSpanId 携带的都是一个来自非 mcpd 的 tracing 系统
的父级,只用于关联。
一个来自别处的 span
// Ids, parent and times the caller already holds. Queued as given.
mcpd.record({
traceId: otel.traceId,
spanId: otel.spanId,
parentSpanId: otel.parentSpanId,
name: 'GET /orders',
kind: 'http_request',
status: 'ok',
startTime: otel.startTime,
endTime: otel.endTime,
isRoot: false,
isEvent: false,
correlation: { project: 'support' },
receivedAt: otel.endTime,
})
// A parent from a tracing system that is not mcpd, for correlation only.
const run = mcpd.trace('handle_request', { externalParentSpanId: otel.spanId }) 你可以依赖的东西
-
接入代码从不抛异常。一次失败的导出、一个不可达的守护进程,或者一个序列化
不了的载荷,都会通过
onError每个间隔最多上报一次,然后就放过 它。可观测性绝不能把它观测的那个东西搞垮。 - 记录不阻塞。它往队列里追加一条就返回;网络工作发生在一个后台 worker 上, 所以你的智能体的延迟,从不取决于 mcpd 快不快,或者活没活着。
-
内存有上限。队列最多放
maxQueue个信号。队列满时最旧的会被 丢弃,同时计入mcpd.stats(),因为一个没有上限的缓冲区,会把 一次网络故障变成一次内存耗尽被杀。 - 采样决定只在根上做一次,然后往下继承。一条只记了一半的 trace,读起来像是 你系统里的一个 bug。
- 还没结束的 span 一样会发出去。一个 span 在开始时排一次队,在结束时再排一 次,所以一次跑到半路崩掉的运行也是看得见的。如果开始那条记录还没发走, 结束那条会替换掉它,所以一个很快的 span 只花一条记录,而不是两条。
-
你的载荷绝不会被改写。键会在
camelCase和守护进程的snake_case之间转换,而这个转换到input、output、attributes、metadata、fields和labels为止。你的提示词到达时,和你 写下的一模一样。
把它读回来
读也是同一个客户端。每个方法对应一个 HTTP 端点,返回由规范生成的类型。读取
失败时会抛出一个 McpdError:一次你等着结果、却答不上来的查询是
个消息,这和一个导不出去的 span 不一样。
检索与分析
// The same client reads. This is the query that finds bad answers: every run
// that a scorer put below the bar.
const suspect = await mcpd.listTraces({
hasScore: 'answer_relevance',
scoreBelow: 0.5,
limit: 10,
})
console.log(`runs scoring under 0.5: ${suspect.traces.length}`)
const recent = await mcpd.listTraces({ limit: 5, sort: 'start_time', order: 'desc' })
for (const trace of recent.traces) {
const seconds = ((trace.durationNs ?? 0) / 1e9).toFixed(2)
console.log(
`${trace.traceId.slice(0, 8)} ${trace.name.padEnd(16)} ${trace.status.padEnd(6)} ` +
`${seconds}s ${trace.spanCount} spans $${(trace.cost?.totalCost ?? 0).toFixed(6)}`,
)
}
// One run, with every span in it, ordered so a waterfall renders directly.
const detail = await mcpd.getTrace(traceId, { include: ['scores', 'feedback'] })
for (const span of detail.spans) {
console.log(` ${span.kind.padEnd(18)} ${span.name}`)
}