Go

Instalación

Go 1.21 o posterior. No hay dependencias de terceros: el transporte es net/http y compress/gzip de la biblioteca estándar, así que añadir mcpd no añade nada más a tu grafo de módulos.

Construir el cliente

Un cliente por proceso. Cada ajuste tiene una opción y una variable de entorno, así que el mismo código corre en desarrollo y en producción sin ninguna edición. mcpd.New() sin ninguna opción funciona contra un 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)

El código gana al entorno, y el entorno gana al valor por defecto.

WithEnabled(false) convierte el registro en una operación vacía que sigue devolviendo spans e ids válidos, así que una prueba puede desactivar mcpd sin cambiar un punto de llamada. Los métodos de lectura siguen funcionando, porque apagar la instrumentación no debería convertir en silencio una consulta en una respuesta vacía.

WithCapturePayloads(false) mantiene los prompts y las completions dentro del proceso: los spans siguen llevando sus tiempos, su estado, su uso, su modelo, su correlación y sus errores, y input y output se descartan antes de llegar a la cola. Un span puede decir lo contrario para sí mismo con WithSpanCapturePayloads, en cualquiera de los dos sentidos.

WithSink cambia adónde van los lotes. Un RecordingSink los guarda en memoria, así que una prueba comprueba lo que registró la instrumentación sin servidor; un ConsoleSink escribe una línea JSON por señal. Un sink que falla se cuenta en Stats() y nunca llega al código que registró las señales. Record encola un span que ya terminó, con sus ids y sus tiempos, tal cual; uno mal formado se rechaza y se cuenta, nunca se devuelve como error. WithExternalParent lleva el padre de otro sistema de trazas solo para correlacionar.

Registrar una ejecución

Start devuelve un contexto nuevo y un span, y defer span.End() lo cierra. Ese par es la forma con ámbito en Go: el defer se ejecuta en un retorno temprano y en un panic, así que ningún camino de control en el que no pensaste puede dejar un span abierto.

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 crea un hijo del span que haya en el contexto, y una raíz cuando no hay ninguno. Trace siempre inicia una raíz, y span.Child(ctx, name) inicia un hijo de un span concreto cuando no lo llevas en un contexto. Aquí nada inventa un segundo mecanismo de propagación: el padre está en el ctx que ya venías pasando.

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

A través de un límite de servicio, Inject escribe un traceparent de W3C en las cabeceras salientes y Extract lee uno de las cabeceras entrantes hacia un contexto. Las cabeceras que no lo traen se devuelven sin cambios, así que una cabecera ilegible inicia una traza nueva en vez de hacer fallar la solicitud que la traía. Se hereda la decisión de muestreo de aguas arriba, para que las dos mitades de una traza coincidan.

// 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 y errores

mcpd calcula el precio a partir de los recuentos de tokens en la ingesta, así que el costo que ves en la traza es el costo que te facturaron. RecordError recibe un error en vez de una cadena, que es lo que garantiza que un fallo registrado lleve un mensaje. Después End registra error como estado en lugar de 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()),
)

Los demás métodos del span son SetInput, SetOutput, SetMetadata, SetAttributes, SetTags, SetStatus, SetSession, SetThread, SetUser, Event y EndWithStatus. Los mismos campos se pueden asignar cuando el span empieza, a través de las opciones With....

Ejecuciones duraderas

El mismo cliente sirve las ejecuciones duraderas del daemon. NewWorkflow construye la definición y guarda los cuerpos de los pasos; un Worker la registra, hace long polling para obtener leases y atiende cada uno en una goroutine, y StartRun, WatchRun, PublishEvent y el resto reflejan la superficie de ejecuciones una a una. Un cuerpo devuelve mcpd.Retryable(err, wait) para que se reintente dentro del presupuesto del paso, y un span iniciado desde el ctx del handler cae bajo su paso en la traza de la ejecución. Lo que gastó un span así es el uso del paso: los recuentos se suman según terminan los spans, gana el último modelo visto, y SetUsage anula los campos que asigna. Al worker se le puede vigilar desde dentro con WorkerOnLease, WorkerOnStepStart, WorkerOnStepEnd, WorkerOnConflict y 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}"})

Qué hace cada construcción, el protocolo de leases y la API de ejecuciones están en Workflows, Workers y Ejecuciones.

Puntuaciones y feedback

Score recibe min y max como argumentos posicionales normales, así que la escala no se puede omitir. Un 0.86 suelto no significa nada hasta que sabes el rango en el que está. El feedback es el juicio de una persona y se mantiene aparte de una puntuación, porque los dos responden preguntas distintas y nunca se deben promediar juntos.

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

Pasa mcpd.TraceTarget(id) o mcpd.SpanTarget(id, spanID) para nombrar lo que se está juzgando, o un mcpd.Target{} vacío para usar el span que haya en el contexto.

Las cinco señales, un solo juego de ids

Log, Metric, Score y Feedback reciben todos el contexto, así que las cinco señales llevan los mismos ids de traza y de span. Eso es lo que permite que un pico en un gráfico abra la ejecución que lo causó.

Flush y apagado

Start y End añaden a una cola en memoria y vuelven; una goroutine en segundo plano agrupa y envía. Flush(ctx) envía lo que está en la cola sin detener nada. Shutdown(ctx) hace flush, detiene el worker y es idempotente, y por eso defer m.Shutdown(ctx) junto a New es la costumbre correcta.

En una función serverless el runtime congela el proceso en cuanto el handler devuelve, y la goroutine del worker no vuelve a correr, así que se pierde el lote que seguía en la cola. Haz flush antes de devolver.

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
}

Con qué puedes contar

• Ningún método que registre algo devuelve un error ni hace panic, tampoco sobre un cliente nil. Una exportación fallida se anota como mucho una vez por intervalo de flush y se descarta. La observabilidad nunca debe tumbar aquello que observa.
• Registrar no bloquea, así que la latencia de tu agente nunca depende de que mcpd sea rápido, ni de que esté en pie.
• La memoria está acotada. La cola guarda WithMaxQueue señales; a partir de ahí se descartan las más antiguas y se cuentan, visibles en Stats() junto a lo que se exportó, lo que se abandonó y lo que Record rechazó. Un búfer sin límite convierte una caída de red en una muerte por falta de memoria.
• La decisión de muestreo se toma una vez, en la raíz, calculando un hash del id de traza, y la hereda cada hijo. Por eso todos los procesos de una flota toman la misma decisión sobre la misma traza.
• Un span sin terminar también se exporta. Los spans largos salen cuando empiezan y se actualizan cuando terminan, así que una ejecución que se cae a mitad de vuelo es visible. Un span que termina antes del siguiente flush se envía una sola vez, ya terminado.
• Un valor que no se puede codificar se descarta y se anota en vez de hacer fallar tu llamada, porque un error de serialización en un payload no debe aparecer como un fallo de tu agente.

Volver a leerlo

Los métodos de lectura reflejan la API HTTP y devuelven los tipos generados. Son los únicos métodos que devuelven errores, porque quien llama está esperando una respuesta.

// 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 una ejecución cubre los filtros por completo. El modelo de datos lista todos los campos que llevan estos tipos.