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

Go

由 context.Context 携带活动的 span,和其他每一个 Go 库的做法一样。没有第三方依赖,也没有哪个记录调用会失败。

安装

Go 1.21 及以上。没有第三方依赖:传输层用的是标准库里的 net/http 和 compress/gzip,所以加上 mcpd 不会往你的模块图里再添别的东西。

go get github.com/mcpd/mcpd/sdks/go

构造客户端

一个进程一个客户端。每一项设置都有一个选项和一个环境变量,所以同一份代码不用 改一个字,就能在开发环境和生产环境里跑。完全不带选项的 mcpd.New() 对着一个本地的守护进程就能用。

连接

m := mcpd.New(
	mcpd.WithEndpoint("http://localhost:4317"),
	mcpd.WithProject("support"),
	mcpd.WithEnvironment("development"),
	mcpd.WithService("support-agent", "1.4.2"),
)
// Shutdown flushes what is still queued and stops the background worker.
// A short-lived process that skips it loses whatever had not been sent.
defer m.Shutdown(ctx)

代码优先于环境变量,环境变量优先于默认值。

选项环境变量默认值
WithEndpointMCPD_ENDPOINThttp://localhost:4317
WithAPIKeyMCPD_API_KEY无
WithProjectMCPD_PROJECTdefault
WithEnvironmentMCPD_ENVIRONMENT无
WithServiceMCPD_SERVICE_NAME、MCPD_SERVICE_VERSION无
WithSampleRateMCPD_SAMPLE1.0
WithBatchSizeMCPD_BATCH_SIZE128
WithFlushIntervalMCPD_FLUSH_INTERVAL_MS2s
WithMaxQueueMCPD_MAX_QUEUE8192
WithTimeoutMCPD_TIMEOUT_MS10s
WithEnabledMCPD_ENABLEDtrue
WithCapturePayloadsMCPD_CAPTURE_PAYLOADStrue
WithLogLevelMCPD_LOG_LEVELdebug

WithEnabled(false) 把记录变成空操作,但仍然返回合法的 span 和 id,所以一个测试不用改任何调用处就能把 mcpd 关掉。读取方法照常工作,因为 关掉接入,不应该悄悄把一次查询变成一个空答案。

WithCapturePayloads(false) 把提示词和补全留在进程里:span 照样 带着它的时间、状态、用量、模型、关联信息和错误,而 input 和 output 在进入队列之前就被丢掉。单个 span 可以用 WithSpanCapturePayloads 说自己不一样,两个方向都可以。

WithSink 换掉批次去的地方。RecordingSink 把它们留在 内存里,于是一个测试不需要服务器就能断言接入代码记下了什么; ConsoleSink 则每个信号写一行 JSON。失败的 sink 会计入 Stats(),绝不会传到记录这些信号的那段代码里。 Record 把一个已经结束的 span 排进队列,连同 id 和时间,按给的 样子来;格式不对的会被拒绝并计数,绝不作为一个错误返回。 WithExternalParent 携带的是来自另一套 tracing 系统的父级, 只用于关联。

记录一次运行

Start 返回一个新的 context 和一个 span, defer span.End() 把它关掉。这一对就是 Go 里带作用域的那种写法: defer 在提前返回时会跑,在 panic 时也会跑,所以一个 span 不会因为某条你没 想到的控制流而一直开着。

带作用域的写法

ctx, span := m.Start(ctx, "retrieve_docs", mcpd.WithSpanKind(mcpd.SpanKindRetrieval))
defer span.End()

// Everything that takes this ctx from here down becomes a child of the span,
// and End is idempotent, so a defer and an explicit call cannot disagree about
// the duration.

记录一次运行

// 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()
}

上下文

Start 在 context 里那个 span 下面开一个子 span,里面没有时就开 一个根。Trace 永远开一个根;当你没有用 context 携带某个 span 时,span.Child(ctx, name) 在那个指定的 span 下面开一个子 span。 这里没有发明第二套传播机制:父级就在你本来就在传的那个 ctx 里。

嵌套

func handle(ctx context.Context, m *mcpd.Client, q string) string {
    ctx, run := m.Start(ctx, "support_agent", mcpd.WithSpanKind(mcpd.SpanKindAgentRun))
    defer run.End()

    // draft is handed the ctx, not the span. That is the whole mechanism.
    return draft(ctx, m, q)
}

func draft(ctx context.Context, m *mcpd.Client, q string) string {
    _, span := m.Start(ctx, "draft_answer", mcpd.WithSpanKind(mcpd.SpanKindModelGeneration))
    defer span.End()
    return "..."
}

跨过服务边界时,Inject 把一个 W3C traceparent 写到 出去的请求头上,Extract 则从进来的请求头里读一个出来放进 context。请求头里没有它时原样返回,所以一个读不出来的头只会开一条新的 trace,而不会让携带它的那个请求失败。上游的采样决定会被继承,于是一条 trace 的两半是一致的。

跨过服务边界

// Outbound: write the active trace context onto the request.
req, _ := http.NewRequestWithContext(ctx, http.MethodPost, url, body)
mcpd.Inject(ctx, req.Header)

// Inbound: read it back and continue the same trace.
func handler(w http.ResponseWriter, r *http.Request) {
    ctx, span := m.Start(mcpd.Extract(r.Context(), r.Header), "handle_request")
    defer span.End()
}

模型、用量与错误

mcpd 在摄取时就是按 token 数定价的,所以 trace 上的成本就是你被计费的那个 成本。RecordError 收的是一个 error,而不是一个 字符串,正是这一点保证了记下来的失败一定带着一条消息。之后 End 记的状态就是 error,而不是 ok。

一个 span 带着什么

span.SetModel("gpt-4o-mini", "openai")
span.SetTokens(812, 96)

// Or the full breakdown, so a cached prompt is priced as a cached prompt.
span.SetUsage(mcpd.TokenUsage{
    InputTokens:  ptr(uint64(812)),
    OutputTokens: ptr(uint64(96)),
    InputDetails: &mcpd.InputTokenDetails{CacheRead: ptr(uint64(640))},
})

// RecordError takes an error, so a recorded failure always has a message.
span.RecordError(err,
    mcpd.ErrorKind("SMTPTemporaryFailure"),
    mcpd.ErrorRetryable(true),
    mcpd.ErrorStack(mcpd.Stack()),
)

span 其余的方法是 SetInput、SetOutput、 SetMetadata、SetAttributes、SetTags、 SetStatus、SetSession、SetThread、 SetUser、Event 和 EndWithStatus。 同样这些字段,也可以在 span 开始时通过那些 With... 选项设置。

持久化运行

同一个客户端也服务守护进程的持久化运行。NewWorkflow 构建定义并 持有那些步骤主体;一个 Worker 注册它,用长轮询要租约,并在一个 goroutine 上逐份服务,而 StartRun、WatchRun、 PublishEvent 以及其余的方法与运行接口一一对应。步骤主体返回 mcpd.Retryable(err, wait),表示要在这个步骤的预算内重试;而从 处理函数的 ctx 开出来的 span,会落在这次运行的 trace 里它自己 那个步骤下面。这样一个 span 花掉的,就是这个步骤的用量:这些计数在 span 结束时累加,看到的最后一个模型胜出,而 SetUsage 会覆盖它设置 的那些字段。这个 worker 可以从内部观察,用的是 WorkerOnLease、 WorkerOnStepStart、WorkerOnStepEnd、 WorkerOnConflict 和 worker.Stats()。

一个工作流

support := mcpd.NewWorkflow("support_ticket").
	Step("classify", func(ctx context.Context, step *mcpd.StepContext) (any, error) {
		var in struct {
			Ticket string `json:"ticket"`
		}
		if err := step.Input(&in); err != nil {
			return nil, err
		}
		kind := "bug"
		if strings.HasPrefix(in.Ticket, "Q") {
			kind = "question"
		}
		return map[string]any{"kind": kind}, nil
	}).
	Agent("draft", func(ctx context.Context, step *mcpd.StepContext) (any, error) {
		// Your own model call. A span started from ctx lands under this step,
		// and what it spent is the step's usage: the daemon prices the sum.
		_, span := m.Start(ctx, "model_call", mcpd.WithModel("claude-sonnet-4", "anthropic"))
		defer span.End()
		span.SetTokens(812, 37)
		return "an answer", nil
	}, mcpd.Input(map[string]any{"kind": "${steps.classify.output.kind}"})).
	WaitFor("approval", "approved", "${input.ticket}", 7*24*time.Hour,
		mcpd.Guard(mcpd.When("steps.classify.output.kind", "eq", "bug"))).
	Step("send", func(ctx context.Context, step *mcpd.StepContext) (any, error) {
		if err := deliver("the draft", step.IdempotencyKey); err != nil {
			return nil, mcpd.Retryable(err, 2*time.Second)
		}
		return map[string]any{"sent": true}, nil
	}, mcpd.Input(map[string]any{"text": "${steps.draft.output}"}), mcpd.AtMostOnce(),
		mcpd.Retry(mcpd.RetryPolicy{
			MaxAttempts: 3, Backoff: mcpd.BackoffExponential, Initial: 500 * time.Millisecond,
		})).
	Output(map[string]any{"sent": "${steps.send.output.sent}"})

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

评分与反馈

Score 把 min 和 max 当作普通的位置 参数收下,所以量程漏不掉。一个光秃秃的 0.86,在你知道它所在的范围之前没有 任何意义。反馈是人的判断,和评分分开存放,因为这两者回答的是不同的问题, 绝不能被平均到一起。

评分与反馈

// A machine evaluation of the run. min and max are required: 0.86 means
// nothing until you know the scale it sits on.
m.Score(ctx, mcpd.TraceTarget(traceID), "answer_relevance", 0.86, 0, 1,
	mcpd.ScoreReason("Answered the billing question and cited the invoice."),
	mcpd.ScoreVersion("2"),
)

// A person's judgement of the same run. Wire this to your product's
// thumbs-up control and it lands on the exact trace that produced the
// answer.
m.Feedback(ctx, mcpd.TraceTarget(traceID), mcpd.FeedbackKindThumbs, "up",
	mcpd.FeedbackComment("Fixed it first time."),
	mcpd.FeedbackAuthor("u-317", "end_user"),
)

传 mcpd.TraceTarget(id) 或 mcpd.SpanTarget(id, spanID) 来指明被评判的是什么,或者传一个 空的 mcpd.Target{},用 context 里的那个 span。

五种信号,一套 id

Log、Metric、Score 和 Feedback 都收下这个 context,所以五种信号带的是同样的 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

trace_id Store Search Export

刷出与关闭

Start 和 End 往内存里的一个队列追加一条然后返回; 一个后台 goroutine 负责攒批和发送。Flush(ctx) 把排着队的东西 发出去,不停下任何东西。Shutdown(ctx) 先刷出,再停掉 worker, 而且是幂等的,这也是为什么在 New 旁边写一句 defer m.Shutdown(ctx) 是个好习惯。

在一个 serverless 函数里,处理函数一返回,运行时就把进程冻住,worker goroutine 再也不会跑,于是还排在队列里的那个批次就丢了。请在返回之前刷出。

一个 serverless 处理函数

func Handler(ctx context.Context, event Event) (Response, error) {
    ctx, run := m.Start(ctx, "answer")
    answer := agent(ctx, event.Body)
    run.SetOutput(answer)
    run.End()

    // The batch is still in memory here. The runtime freezes this process the
    // moment the handler returns, and the worker goroutine never runs again.
    if err := m.Flush(ctx); err != nil {
        log.Printf("mcpd flush: %v", err)
    }
    return Response{Body: answer}, nil
}

你可以依赖的东西

  • 任何记录东西的方法都不返回错误,也不 panic,在一个 nil 客户端上调用也是 如此。一次失败的导出,每个刷出间隔最多记一条日志,然后丢掉。可观测性绝不 能把它观测的那个东西搞垮。
  • 记录不阻塞,所以你的智能体的延迟,从不取决于 mcpd 快不快,或者活没活着。
  • 内存有上限。队列最多放 WithMaxQueue 个信号;超过之后最旧的 会被丢弃并计数,在 Stats() 里看得到,旁边还有导出了多少、 放弃了多少,以及 Record 拒了多少。一个没有上限的缓冲区,会把 一次网络故障变成一次内存耗尽被杀。
  • 采样决定只在根上做一次,做法是对 trace id 取哈希,然后由每个子级继承。 因此一个集群里的每个进程,对同一条 trace 都会做出同样的决定。
  • 没结束的 span 一样会导出。长时间运行的 span 在开始时就发出去,在结束时 更新,所以一次跑到半路崩掉的运行是看得见的。在下一次刷出之前就结束的 span 只发一次,发的是它结束后的样子。
  • 编码不了的值会被丢弃并记进日志,而不是让你的调用失败,因为一个载荷里的 序列化错误,绝不能冒出来变成你的智能体里的一次失败。

把它读回来

读取方法对应 HTTP API,返回由规范生成的类型。它们是唯一一组会返回错误的 方法,因为调用方正等着一个答案。

检索与分析

// Recording is asynchronous, so wait for the batch to land before reading
// it back. Outside an example you would never need this.
if err := m.Flush(ctx); err != nil {
	log.Printf("flush: %v", err)
}

detail, err := m.GetTrace(ctx, traceID, "scores", "feedback")
if err != nil {
	log.Fatalf("read the run back: %v", err)
}
fmt.Printf("%s: %s in %s, %d spans\n",
	detail.Trace.TraceID, detail.Trace.Status, duration(detail.Trace.DurationNs), len(detail.Spans))
for _, span := range detail.Spans {
	fmt.Printf("  %-18s %-18s %s\n", span.Kind, span.Name, span.Status)
}

summary, err := m.AnalyticsSummary(ctx, time.Now().Add(-time.Hour), time.Now(), "model")
if err != nil {
	log.Fatalf("read the summary: %v", err)
}
fmt.Printf("last hour: %d runs, %.1f%% failed, %.4f %s\n",
	summary.TraceCount, summary.ErrorRate*100, summary.TotalCost, summary.Currency)

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