Go
다른 모든 Go 라이브러리가 하는 방식 그대로, context.Context가 활성 스팬을 나릅니다. 서드파티 의존성이 없고, 실패할 수 있는 기록 호출도 없습니다.
설치
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) 코드가 환경 변수를 이기고, 환경 변수가 기본값을 이깁니다.
| 옵션 | 환경 변수 | 기본값 |
|---|---|---|
WithEndpoint | MCPD_ENDPOINT | http://localhost:4317 |
WithAPIKey | MCPD_API_KEY | 없음 |
WithProject | MCPD_PROJECT | default |
WithEnvironment | MCPD_ENVIRONMENT | 없음 |
WithService | MCPD_SERVICE_NAME, MCPD_SERVICE_VERSION | 없음 |
WithSampleRate | MCPD_SAMPLE | 1.0 |
WithBatchSize | MCPD_BATCH_SIZE | 128 |
WithFlushInterval | MCPD_FLUSH_INTERVAL_MS | 2s |
WithMaxQueue | MCPD_MAX_QUEUE | 8192 |
WithTimeout | MCPD_TIMEOUT_MS | 10s |
WithEnabled | MCPD_ENABLED | true |
WithCapturePayloads | MCPD_CAPTURE_PAYLOADS | true |
WithLogLevel | MCPD_LOG_LEVEL | debug |
WithEnabled(false)는 기록을 아무 일도 하지 않는 동작으로 바꾸면서도
올바른 스팬과 id는 그대로 돌려줍니다. 그래서 테스트가 호출 지점을 바꾸지 않고 mcpd를
끌 수 있습니다. 읽기 메서드는 그대로 동작합니다. 계측을 껐다고 해서 쿼리가 소리 없이
빈 답으로 바뀌어서는 안 되기 때문입니다.
WithCapturePayloads(false)는 프롬프트와 컴플리션을 프로세스 안에 붙들어
둡니다. 스팬은 시각과 상태와 사용량과 모델과 상관 정보와 오류를 그대로 싣고,
input과 output은 큐에 닿기 전에 버려집니다. 스팬 하나는
WithSpanCapturePayloads로 자기 몫만 어느 방향으로든 달리 정할 수
있습니다.
WithSink는 배치가 가는 곳을 바꿉니다. RecordingSink는
배치를 메모리에 들고 있어서, 테스트가 서버 없이 계측이 무엇을 기록했는지 검증합니다.
ConsoleSink는 시그널마다 JSON 한 줄을 씁니다. 실패한 싱크는
Stats()에 세고, 시그널을 기록한 코드에는 결코 닿지 않습니다.
Record는 이미 끝난 스팬을 id와 시각까지 준 그대로 큐에 넣습니다. 형식이
깨진 것은 거절하고 세며, 오류로 돌려주지는 않습니다.
WithExternalParent는 다른 트레이싱 시스템의 부모를 상관 정보로만 실어
나릅니다.
실행 기록하기
Start는 새 컨텍스트와 스팬을 돌려주고, defer span.End()가
그것을 닫습니다. Go에서는 이 짝이 범위를 가진 형태입니다. defer는 이른 반환에서도
패닉에서도 돌아가므로, 미처 생각하지 못한 제어 흐름 때문에 스팬이 열린 채 남는 일이
없습니다.
범위를 가진 형태
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는 컨텍스트에 들어 있는 스팬의 자식을 만들고, 아무것도 없으면
루트를 만듭니다. Trace는 언제나 루트를 시작합니다. 스팬을 컨텍스트에
담아 나르고 있지 않을 때는 span.Child(ctx, name)이 특정 스팬의 자식을
시작합니다. 여기에 두 번째 전파 방식을 새로 만들어 넣은 것은 없습니다. 부모는 이미
넘기고 있던 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가 들어온 헤더에서 그것을 읽어
컨텍스트에 넣습니다. 그것이 없는 헤더는 바뀌지 않은 채로 돌아오므로, 읽을 수 없는
헤더는 그것을 실어 온 요청을 실패시키는 대신 새 트레이스를 시작합니다. 상류의 샘플링
결정을 물려받으므로, 트레이스의 양쪽 절반이 서로 어긋나지 않습니다.
서비스 경계를 넘어
// 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가 수집 시점에 값을 매기는 대상이 토큰 수이므로, 트레이스에 적힌 비용이 곧
청구받은 비용입니다. RecordError는 문자열이 아니라 error를
받습니다. 기록된 실패에 메시지가 붙어 있음을 보장하는 것이 바로 이것입니다. 그러면
End는 상태를 ok가 아니라 error로 기록합니다.
스팬이 싣는 것
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()),
)
나머지 스팬 메서드는 SetInput, SetOutput,
SetMetadata, SetAttributes, SetTags,
SetStatus, SetSession, SetThread,
SetUser, Event, EndWithStatus입니다. 같은
필드는 With... 옵션으로 스팬이 시작할 때 정할 수도 있습니다.
지속성 실행
같은 클라이언트가 데몬의 지속성 실행도 다룹니다. NewWorkflow는 정의를
만들고 단계 본문을 들고 있습니다. Worker는 그 정의를 등록하고, 리스를
롱 폴링으로 기다렸다가 고루틴 위에서 하나씩 처리합니다. StartRun,
WatchRun, PublishEvent와 나머지는 실행 표면을 하나씩 그대로
옮겨 놓은 것입니다. 본문은 단계의 예산 안에서 다시 시도되도록
mcpd.Retryable(err, wait)을 돌려주고, 핸들러의 ctx에서
시작한 스팬은 실행의 트레이스에서 그 단계 아래에 놓입니다. 그런 스팬이 쓴 것이 곧 그
단계의 사용량입니다. 수치는 스팬이 끝날 때마다 더해지고, 마지막으로 본 모델이
이기며, SetUsage는 자기가 정한 필드를 덮어씁니다. 워커는
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는 워크플로, 워커, 실행에 있습니다.
점수와 피드백
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{}을 넘기십시오.
다섯 시그널, 하나의 id
Log, Metric, Score,
Feedback은 모두 컨텍스트를 받으므로, 다섯 시그널이 같은 트레이스 id와
스팬 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
플러시와 종료
Start와 End는 메모리 안의 큐에 덧붙이고 바로 돌아옵니다.
배치로 묶어 보내는 일은 백그라운드 고루틴이 합니다. Flush(ctx)는
아무것도 멈추지 않고 큐에 있는 것을 보냅니다. Shutdown(ctx)은
플러시하고, 워커를 멈추며, 멱등합니다. New 옆에
defer m.Shutdown(ctx)를 두는 것이 옳은 습관인 이유가 그것입니다.
서버리스 함수에서는 런타임이 핸들러가 돌아오는 순간 프로세스를 얼리고 워커 고루틴은 다시 돌지 않으므로, 아직 큐에 남아 있던 배치를 잃습니다. 돌아가기 전에 플러시하십시오.
서버리스 핸들러
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
} 믿어도 되는 것
- 무언가를 기록하는 메서드는 오류를 돌려주지도, 패닉을 일으키지도 않습니다. nil 클라이언트에서도 그렇습니다. 실패한 내보내기는 플러시 주기에 많아야 한 번 로그로 남기고 버립니다. 관측 가능성이 관측 대상을 무너뜨리는 일은 절대 없어야 합니다.
- 기록은 블로킹하지 않으므로, 에이전트의 지연 시간이 mcpd가 빠른지, 살아 있는지에 달리는 일이 없습니다.
-
메모리에는 상한이 있습니다. 큐는
WithMaxQueue개의 시그널을 담고, 그 너머로는 가장 오래된 것부터 버리고 셉니다. 그 수는 내보낸 것, 포기한 것,Record가 거절한 것과 함께Stats()에서 보입니다. 상한 없는 버퍼는 네트워크 장애를 메모리 부족으로 인한 강제 종료로 바꿔 놓습니다. - 샘플링 결정은 루트에서 트레이스 id를 해싱해 한 번 내리고, 모든 자식이 물려받습니다. 그래서 한 무리의 모든 프로세스가 같은 트레이스에 대해 같은 결정을 내립니다.
- 끝나지 않은 스팬도 내보냅니다. 오래 도는 스팬은 시작할 때 나가고 끝날 때 갱신되므로, 도중에 죽은 실행도 보입니다. 다음 플러시 전에 끝난 스팬은 끝난 채로 한 번만 나갑니다.
- 인코딩되지 않는 값은 호출을 실패시키는 대신 버리고 로그로 남깁니다. 페이로드의 직렬화 오류가 에이전트의 실패로 드러나서는 안 되기 때문입니다.
다시 읽기
읽기 메서드는 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)