Go

Instalação

Go 1.21 ou posterior. Não há nenhuma dependência de terceiros: o transporte é net/http e compress/gzip da biblioteca padrão, então acrescentar o mcpd não acrescenta mais nada ao grafo do seu módulo.

Construir o cliente

Um cliente por processo. Todo ajuste tem uma opção e uma variável de ambiente, então o mesmo código roda em desenvolvimento e em produção sem nenhuma edição. mcpd.New() sem opção nenhuma funciona contra um daemon local.

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)

O código vence o ambiente, e o ambiente vence o padrão.

WithEnabled(false) transforma o registro em uma operação vazia que ainda devolve spans e ids válidos, então um teste pode desligar o mcpd sem mudar um ponto de chamada. Os métodos de leitura continuam funcionando, porque desligar a instrumentação não deve transformar em silêncio uma consulta em uma resposta vazia.

WithCapturePayloads(false) mantém prompts e respostas dentro do processo: os spans continuam carregando o seu tempo, o seu status, o seu uso, o seu modelo, a sua correlação e os seus erros, e input e output são descartados antes de chegar à fila. Um span pode dizer o contrário para si mesmo com WithSpanCapturePayloads, nos dois sentidos.

WithSink troca o destino dos lotes. Um RecordingSink os guarda em memória, então um teste verifica o que a instrumentação registrou sem nenhum servidor; um ConsoleSink escreve uma linha JSON por sinal. Um sink que falha é contado em Stats() e nunca chega ao código que registrou os sinais. Record enfileira um span que já terminou, com ids e tempos incluídos, como veio; um malformado é recusado e contado, nunca devolvido como erro. WithExternalParent carrega o pai de outro sistema de tracing, só para correlação.

Registrar uma execução

Start devolve um contexto novo e um span, e defer span.End() o fecha. Esse par é a forma com escopo em Go: o defer roda em um retorno antecipado e em um panic, então um span não pode ficar aberto por um caminho de fluxo em que você não pensou.

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

Contexto

Start cria um filho de qualquer span que estiver no contexto, e uma raiz quando não houver nenhum. Trace sempre começa uma raiz, e span.Child(ctx, name) começa um filho de um span específico quando você não o está carregando em um contexto. Nada aqui inventa um segundo mecanismo de propagação: o pai está no ctx que você já vinha passando.

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 "..."
}

Através de uma fronteira de serviço, Inject escreve um traceparent do W3C nos cabeçalhos de saída e Extract lê um dos cabeçalhos de entrada para dentro de um contexto. Cabeçalhos sem ele voltam sem alteração, então um cabeçalho ilegível começa um trace novo em vez de falhar a requisição que o carregou. A decisão de amostragem de quem está acima é herdada, então as duas metades de um trace concordam.

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

Modelo, uso e erros

As contagens de tokens são o que o mcpd precifica na ingestão, então o custo no trace é o custo que foi cobrado de você. RecordError recebe um error em vez de uma string, e é isso que garante que uma falha registrada tenha uma mensagem junto. O End então registra error como status, e não ok.

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()),
)

Os outros métodos do span são SetInput, SetOutput, SetMetadata, SetAttributes, SetTags, SetStatus, SetSession, SetThread, SetUser, Event e EndWithStatus. Os mesmos campos podem ser definidos quando o span começa, através das opções With....

Execuções duráveis

O mesmo cliente serve as execuções duráveis do daemon. NewWorkflow constrói a definição e guarda os corpos das etapas; um Worker a registra, faz long polling por leases e serve cada um em uma goroutine, e StartRun, WatchRun, PublishEvent e os demais espelham a superfície de execuções uma a uma. Um corpo devolve mcpd.Retryable(err, wait) para ser repetido dentro do orçamento da etapa, e um span iniciado a partir do ctx do handler cai abaixo da sua etapa no trace da execução. O que um span desses gastou é o uso da etapa: as contagens são somadas conforme os spans terminam, o último modelo visto vence, e SetUsage sobrescreve os campos que definir. Dá para acompanhar o worker por dentro através de WorkerOnLease, WorkerOnStepStart, WorkerOnStepEnd, WorkerOnConflict e 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}"})

O que cada construção faz, o protocolo de lease e a API de execuções estão em Workflows, Workers e Execuções.

Notas e feedback

Score recebe min e max como argumentos posicionais comuns, então a escala não pode ficar de fora. Um 0.86 solto não quer dizer nada enquanto você não souber o intervalo em que ele está. O feedback é o julgamento de uma pessoa e fica separado de uma nota, porque os dois respondem perguntas diferentes e nunca devem entrar na mesma média.

// 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"),
)

Passe mcpd.TraceTarget(id) ou mcpd.SpanTarget(id, spanID) para nomear o que está sendo julgado, ou um mcpd.Target{} vazio para usar o span que estiver no contexto.

Os cinco sinais, um único conjunto de ids

Log, Metric, Score e Feedback recebem todos o contexto, então os cinco sinais carregam os mesmos ids de trace e de span. É isso que permite a um pico em um gráfico abrir a execução que o causou.

Flush e encerramento

Start e End acrescentam a uma fila em memória e retornam; uma goroutine em segundo plano agrupa em lotes e envia. Flush(ctx) envia o que está na fila sem parar nada. Shutdown(ctx) faz o flush, para o worker e é idempotente, e é por isso que defer m.Shutdown(ctx) ao lado de New é o hábito certo.

Em uma função serverless o runtime congela o processo no instante em que o handler retorna, e a goroutine do worker nunca mais roda, então um lote que ainda estava na fila se perde. Faça o flush antes de retornar.

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
}

Com o que você pode contar

• Nenhum método que registra alguma coisa devolve erro ou entra em panic, nem mesmo em um cliente nil. Uma exportação que falha é logada no máximo uma vez por intervalo de flush e descartada. A observabilidade nunca pode derrubar aquilo que ela observa.
• Registrar não bloqueia, então a latência do seu agente nunca depende de o mcpd ser rápido, nem de estar no ar.
• A memória é limitada. A fila guarda WithMaxQueue sinais; passando disso, os mais antigos são descartados e contados, visíveis em Stats() ao lado do que foi exportado, do que foi abandonado e do que o Record recusou. Um buffer sem limite transforma uma queda de rede em uma morte por falta de memória.
• A decisão de amostragem é tomada uma vez, na raiz, aplicando um hash ao id do trace, e herdada por todo filho. Todo processo de uma frota toma então a mesma decisão sobre o mesmo trace.
• Um span inacabado também é exportado. Spans longos saem quando começam e são atualizados quando terminam, então uma execução que quebra no meio do caminho fica visível. Um span que termina antes do próximo flush é enviado uma vez só, já concluído.
• Um valor que não codifica é descartado e logado em vez de fazer a sua chamada falhar, porque um erro de serialização em um payload não pode aparecer como uma falha no seu agente.

Ler de volta

Os métodos de leitura espelham a API HTTP e devolvem os tipos gerados. São os únicos métodos que devolvem erros, porque quem chamou está esperando uma resposta.

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

Encontrar uma execução cobre os filtros por completo. O modelo de dados lista cada campo que esses tipos carregam.