Go

Installazione

Go 1.21 o successivo. Non ci sono dipendenze di terze parti: il trasporto è net/http e compress/gzip della libreria standard, quindi aggiungere mcpd non aggiunge nient’altro al grafo dei tuoi moduli.

Costruire il client

Un client per processo. Ogni impostazione ha un’opzione e una variabile d’ambiente, così lo stesso codice gira in sviluppo e in produzione senza una modifica. mcpd.New() senza nessuna opzione funziona contro un daemon locale.

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)

Il codice vince sull’ambiente, e l’ambiente vince sull’impostazione predefinita.

WithEnabled(false) trasforma la registrazione in un’operazione a vuoto che restituisce comunque span e identificatori validi, così un test può spegnere mcpd senza cambiare un punto di chiamata. I metodi di lettura continuano a funzionare, perché spegnere la strumentazione non deve trasformare di nascosto una query in una risposta vuota.

WithCapturePayloads(false) tiene i prompt e le completion dentro il processo: gli span portano lo stesso i loro tempi, lo stato, il consumo, il modello, la correlazione e gli errori, e input e output vengono scartati prima di arrivare alla coda. Un singolo span può dire il contrario per sé con WithSpanCapturePayloads, in tutte e due le direzioni.

WithSink cambia dove finiscono i batch. Un RecordingSink li tiene in memoria, così un test verifica che cosa ha registrato la strumentazione senza un server; un ConsoleSink scrive una riga JSON per segnale. Un sink che fallisce viene contato in Stats() e non arriva mai al codice che ha registrato i segnali. Record mette in coda uno span già concluso, identificatori e tempi compresi, così com’è; uno malformato viene rifiutato e contato, mai restituito come errore. WithExternalParent porta il padre che viene da un altro sistema di tracing solo per la correlazione.

Registrare un’esecuzione

Start restituisce un contesto nuovo e uno span, e defer span.End() lo chiude. Quella coppia è la forma con ambito in Go: il defer gira su un ritorno anticipato e su un panic, quindi uno span non può restare aperto per un percorso di controllo a cui non avevi pensato.

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

// Tutto quello che prende questo ctx da qui in giù diventa figlio dello span,
// ed End è idempotente, quindi un defer e una chiamata esplicita non possono
// contraddirsi sulla durata.

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

Contesto

Start fa un figlio di qualunque span sia nel contesto, e una radice quando non ce n’è nessuno. Trace avvia sempre una radice, e span.Child(ctx, name) avvia il figlio di uno span preciso quando non lo stai portando in un contesto. Qui non si inventa un secondo meccanismo di propagazione: il padre sta nel ctx che stavi già 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()

    // A draft si passa il ctx, non lo span. Il meccanismo è tutto qui.
    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 "..."
}

Attraverso il confine fra due servizi, Inject scrive un traceparent W3C sugli header in uscita ed Extract ne legge uno dagli header in entrata dentro un contesto. Gli header che non ne portano uno tornano indietro immutati, così un header illeggibile fa partire una traccia nuova anziché far fallire la richiesta che lo portava. La decisione di campionamento presa a monte viene ereditata, così le due metà di una traccia vanno d’accordo.

// In uscita: scrivi il contesto di traccia attivo sulla richiesta.
req, _ := http.NewRequestWithContext(ctx, http.MethodPost, url, body)
mcpd.Inject(ctx, req.Header)

// In entrata: rileggilo e continua la stessa traccia.
func handler(w http.ResponseWriter, r *http.Request) {
    ctx, span := m.Start(mcpd.Extract(r.Context(), r.Header), "handle_request")
    defer span.End()
}

Modello, consumo ed errori

I conteggi dei token sono quelli che mcpd valorizza all’ingest, quindi il costo che vedi sulla traccia è il costo che ti è stato fatturato. RecordError prende un error anziché una stringa, ed è questo a garantire che un guasto registrato porti un messaggio con sé. End registra poi error come stato anziché ok.

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

// Oppure il dettaglio completo, così un prompt in cache è valorizzato come tale.
span.SetUsage(mcpd.TokenUsage{
    InputTokens:  ptr(uint64(812)),
    OutputTokens: ptr(uint64(96)),
    InputDetails: &mcpd.InputTokenDetails{CacheRead: ptr(uint64(640))},
})

// RecordError prende un error, così un guasto registrato ha sempre un messaggio.
span.RecordError(err,
    mcpd.ErrorKind("SMTPTemporaryFailure"),
    mcpd.ErrorRetryable(true),
    mcpd.ErrorStack(mcpd.Stack()),
)

Gli altri metodi dello span sono SetInput, SetOutput, SetMetadata, SetAttributes, SetTags, SetStatus, SetSession, SetThread, SetUser, Event ed EndWithStatus. Gli stessi campi si possono impostare quando lo span parte, con le opzioni With....

Esecuzioni durevoli

Lo stesso client serve le esecuzioni durevoli del daemon. NewWorkflow costruisce la definizione e conserva i corpi dei passi; un Worker la registra, fa long polling per ottenere i lease e serve ognuno di essi su una goroutine, e StartRun, WatchRun, PublishEvent e gli altri rispecchiano uno a uno la superficie delle esecuzioni. Un corpo restituisce mcpd.Retryable(err, wait) per essere ritentato entro il budget del suo passo, e uno span avviato a partire dal ctx dell’handler finisce sotto il proprio passo nella traccia dell’esecuzione. Quello che uno span del genere ha speso è il consumo del passo: i conteggi vengono sommati man mano che gli span finiscono, vince l’ultimo modello visto, e SetUsage sovrascrive i campi che imposta. Il worker si può osservare da dentro con 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}"})

Che cosa fa ogni costrutto, il protocollo dei lease e l’API delle esecuzioni stanno su Workflow, Worker ed Esecuzioni.

Punteggi e feedback

Score prende min e max come normali argomenti posizionali, quindi la scala non si può lasciare fuori. Un 0.86 nudo non vuol dire niente finché non sai su che intervallo sta. Il feedback è il giudizio di una persona e resta separato da un punteggio, perché i due rispondono a domande diverse e non vanno mai mediati insieme.

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

Passa mcpd.TraceTarget(id) o mcpd.SpanTarget(id, spanID) per nominare che cosa stai giudicando, oppure un mcpd.Target{} vuoto per prendere lo span che sta nel contesto.

Tutti e cinque i segnali, un solo insieme di id

Log, Metric, Score e Feedback prendono tutti il contesto, quindi tutti e cinque i segnali portano gli stessi id di traccia e di span. È questo che permette a un picco su un grafico di aprire l’esecuzione che lo ha causato.

Flush e arresto

Start ed End aggiungono a una coda in memoria e ritornano; una goroutine in secondo piano raggruppa in batch e manda. Flush(ctx) manda quello che è in coda senza fermare niente. Shutdown(ctx) fa il flush, ferma il worker ed è idempotente, ed è per questo che defer m.Shutdown(ctx) accanto a New è la buona abitudine.

In una funzione serverless il runtime congela il processo nel momento in cui l’handler ritorna, e la goroutine del worker non gira mai più, quindi un batch ancora in coda è perso. Fai il flush prima di ritornare.

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

    // Qui il batch è ancora in memoria. Il runtime congela il processo nel
    // momento in cui l'handler ritorna, e la goroutine del worker non gira più.
    if err := m.Flush(ctx); err != nil {
        log.Printf("mcpd flush: %v", err)
    }
    return Response{Body: answer}, nil
}

Su che cosa puoi contare

• Nessun metodo che registra qualcosa restituisce un errore o va in panic, nemmeno su un client nil. Un export fallito finisce in un log al massimo una volta per intervallo di flush e poi viene scartato. L’osservabilità non deve mai far cadere la cosa che osserva.
• Registrare non blocca, quindi la latenza del tuo agente non dipende mai da quanto mcpd sia veloce, né dal fatto che sia attivo.
• La memoria è limitata. La coda tiene WithMaxQueue segnali; oltre quel punto i più vecchi vengono scartati e contati, e si vedono in Stats() accanto a quello che è stato esportato, a quello su cui si è rinunciato e a quello che Record ha rifiutato. Un buffer senza limite trasforma un’interruzione di rete in un processo ucciso per memoria esaurita.
• La decisione di campionamento si prende una volta sola, alla radice, con un hash dell’id di traccia, e la ereditano tutti i figli. Ogni processo di una flotta prende quindi la stessa decisione sulla stessa traccia.
• Uno span non finito viene esportato lo stesso. Gli span lunghi escono quando partono e vengono aggiornati quando finiscono, così un’esecuzione che va in crash per strada resta visibile. Uno span che finisce prima del flush successivo viene mandato una volta sola, già concluso.
• Un valore che non si lascia codificare viene scartato e finisce in un log anziché far fallire la tua chiamata, perché un errore di serializzazione dentro un payload non deve affiorare come un guasto nel tuo agente.

Rileggerla

I metodi di lettura rispecchiano l’API HTTP e restituiscono i tipi generati. Sono gli unici metodi che restituiscono errori, perché chi chiama sta aspettando una risposta.

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

Trovare un’esecuzione copre i filtri per intero. Il modello dei dati elenca ogni campo che questi tipi portano.