Go

Installation

Go 1.21 ou plus récent. Il n’y a aucune dépendance tierce : le transport est net/http et compress/gzip de la bibliothèque standard, si bien qu’ajouter mcpd n’ajoute rien d’autre à votre graphe de modules.

Construire le client

Un client par processus. Chaque réglage a une option et une variable d’environnement, si bien que le même code tourne en développement et en production sans la moindre modification. mcpd.New(), sans aucune option, fonctionne face à 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)

Le code l’emporte sur l’environnement, et l’environnement l’emporte sur la valeur par défaut.

WithEnabled(false) transforme l’enregistrement en une opération sans effet qui renvoie quand même des spans et des identifiants valides : un test peut donc désactiver mcpd sans modifier un seul site d’appel. Les méthodes de lecture continuent de fonctionner, car couper l’instrumentation ne doit pas transformer discrètement une requête en réponse vide.

WithCapturePayloads(false) garde les prompts et les complétions dans le processus : les spans portent toujours leurs temps, leur statut, leur consommation, leur modèle, leur corrélation et leurs erreurs, et input et output sont écartés avant d’atteindre la file d’attente. Un span peut décider autrement pour lui-même avec WithSpanCapturePayloads, dans un sens comme dans l’autre.

WithSink change la destination des lots. Un RecordingSink les garde en mémoire : un test vérifie ainsi ce que l’instrumentation a enregistré sans serveur. Un ConsoleSink écrit une ligne JSON par signal. Un sink qui échoue est compté dans Stats() et n’atteint jamais le code qui a enregistré les signaux. Record met en file un span déjà terminé, identifiants et temps compris, tel quel ; un span mal formé est refusé et compté, jamais renvoyé comme une erreur. WithExternalParent porte le parent venu d’un autre système de tracing, pour la seule corrélation.

Enregistrer une exécution

Start renvoie un nouveau contexte et un span, et defer span.End() le referme. Cette paire est la forme à portée délimitée en Go : le defer s’exécute sur un retour anticipé comme sur une panique, si bien qu’un span ne peut pas rester ouvert à cause d’un chemin d’exécution auquel vous n’aviez pas pensé.

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

// Tout ce qui prend ce ctx à partir d’ici devient un enfant du span, et End est
// idempotent, si bien qu’un defer et un appel explicite ne peuvent pas se
// contredire sur la durée.

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

Contexte

Start fait un enfant du span présent dans le contexte, et une racine quand il n’y en a pas. Trace démarre toujours une racine, et span.Child(ctx, name) démarre un enfant d’un span précis quand vous ne le transportez pas dans un contexte. Rien ici n’invente un second mécanisme de propagation : le parent est dans le ctx que vous passiez déjà.

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 reçoit le ctx, pas le span. Tout le mécanisme est là.
    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 "..."
}

D’un service à l’autre, Inject écrit un traceparent W3C sur les en-têtes sortants et Extract en lit un depuis les en-têtes entrants vers un contexte. Des en-têtes qui n’en portent pas sont renvoyés tels quels : un en-tête illisible démarre donc une nouvelle trace au lieu de faire échouer la requête qui le transportait. La décision d’échantillonnage amont est héritée, si bien que les deux moitiés d’une trace s’accordent.

// En sortie, écrivez le contexte de trace actif sur la requête.
req, _ := http.NewRequestWithContext(ctx, http.MethodPost, url, body)
mcpd.Inject(ctx, req.Header)

// En entrée, relisez-le et continuez la même trace.
func handler(w http.ResponseWriter, r *http.Request) {
    ctx, span := m.Start(mcpd.Extract(r.Context(), r.Header), "handle_request")
    defer span.End()
}

Modèle, consommation et erreurs

mcpd tarife les décomptes de tokens à l’ingestion : le coût porté par la trace est donc celui qui vous a été facturé. RecordError prend une error plutôt qu’une chaîne, ce qui garantit qu’un échec enregistré porte un message. End inscrit ensuite error comme statut, et non ok.

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

// Ou le détail complet, pour qu’un prompt en cache soit tarifé comme tel.
span.SetUsage(mcpd.TokenUsage{
    InputTokens:  ptr(uint64(812)),
    OutputTokens: ptr(uint64(96)),
    InputDetails: &mcpd.InputTokenDetails{CacheRead: ptr(uint64(640))},
})

// RecordError prend une error, donc un échec enregistré porte toujours un message.
span.RecordError(err,
    mcpd.ErrorKind("SMTPTemporaryFailure"),
    mcpd.ErrorRetryable(true),
    mcpd.ErrorStack(mcpd.Stack()),
)

Les autres méthodes du span sont SetInput, SetOutput, SetMetadata, SetAttributes, SetTags, SetStatus, SetSession, SetThread, SetUser, Event et EndWithStatus. Les mêmes champs peuvent être renseignés au démarrage du span par les options With....

Exécutions durables

Le même client dessert les exécutions durables du daemon. NewWorkflow construit la définition et retient le corps des étapes ; un Worker l’enregistre, attend des leases en long polling et sert chacun d’eux sur une goroutine, et StartRun, WatchRun, PublishEvent et les autres reprennent un pour un la surface des exécutions. Un corps d’étape renvoie mcpd.Retryable(err, wait) pour être réessayé dans le budget de l’étape, et un span démarré depuis le ctx du handler se place sous son étape dans la trace de l’exécution. Ce qu’un tel span a dépensé est la consommation de l’étape : les décomptes sont additionnés à la fin des spans, le dernier modèle vu l’emporte, et SetUsage remplace les champs qu’il renseigne. Le worker s’observe de l’intérieur avec WorkerOnLease, WorkerOnStepStart, WorkerOnStepEnd, WorkerOnConflict et 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}"})

Ce que fait chaque construction, le protocole de lease et l’API des exécutions sont décrits dans Workflows, Workers et Exécutions.

Scores et feedback

Score prend min et max comme de simples arguments positionnels : l’échelle ne peut donc pas être omise. Un 0.86 tout nu ne veut rien dire tant que vous ignorez l’intervalle sur lequel il se place. Le feedback est le jugement d’une personne et reste distinct d’un score, car les deux répondent à des questions différentes et ne doivent jamais être moyennés ensemble.

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

Passez mcpd.TraceTarget(id) ou mcpd.SpanTarget(id, spanID) pour nommer ce qui est jugé, ou un mcpd.Target{} vide pour prendre le span présent dans le contexte.

Les cinq signaux, un seul jeu d’identifiants

Log, Metric, Score et Feedback prennent tous le contexte, si bien que les cinq signaux portent les mêmes identifiants de trace et de span. C’est ce qui permet à un pic sur un graphique d’ouvrir l’exécution qui l’a causé.

Flush et arrêt

Start et End ajoutent à une file d’attente en mémoire et rendent la main ; une goroutine de fond regroupe et envoie. Flush(ctx) envoie ce qui est en file sans rien arrêter. Shutdown(ctx) vide la file, arrête le worker et est idempotent, ce qui fait de defer m.Shutdown(ctx) à côté de New la bonne habitude.

Dans une fonction serverless, le runtime gèle le processus dès que le handler rend la main, et la goroutine de fond ne tourne plus jamais : un lot encore en file est donc perdu. Videz la file avant de rendre la main.

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

    // Le lot est encore en mémoire ici. Le runtime gèle ce processus dès que le
    // handler rend la main, et la goroutine de fond ne tourne plus jamais.
    if err := m.Flush(ctx); err != nil {
        log.Printf("mcpd flush: %v", err)
    }
    return Response{Body: answer}, nil
}

Ce sur quoi vous pouvez compter

• Aucune méthode qui enregistre quoi que ce soit ne renvoie d’erreur ni ne panique, y compris sur un client nil. Un export en échec est journalisé au plus une fois par intervalle de flush, puis abandonné. L’observabilité ne doit jamais faire tomber ce qu’elle observe.
• L’enregistrement ne bloque pas : la latence de votre agent ne dépend donc jamais de la vitesse de mcpd, ni de sa disponibilité.
• La mémoire est bornée. La file contient WithMaxQueue signaux ; au-delà, les plus anciens sont écartés et comptés, visibles dans Stats() à côté de ce qui a été exporté, de ce qui a été abandonné et de ce que Record a refusé. Un tampon sans limite transforme une panne réseau en arrêt pour mémoire épuisée.
• La décision d’échantillonnage est prise une seule fois, à la racine, en hachant l’identifiant de trace, puis héritée par chaque enfant. Tous les processus d’une flotte prennent donc la même décision sur la même trace.
• Un span inachevé est exporté quand même. Les spans de longue durée partent à leur démarrage et sont mis à jour à leur fin, si bien qu’une exécution qui plante en plein vol reste visible. Un span qui se termine avant le flush suivant est envoyé une seule fois, terminé.
• Une valeur qui refuse de s’encoder est écartée et journalisée plutôt que de faire échouer votre appel, car une erreur de sérialisation dans un payload ne doit pas ressortir comme une panne de votre agent.

Relire ce qui a été enregistré

Les méthodes de lecture reflètent l’API HTTP et renvoient les types générés. Ce sont les seules méthodes qui renvoient des erreurs, parce que l’appelant attend une réponse.

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

Retrouver une exécution couvre les filtres en détail. Le modèle de données liste chaque champ que ces types portent.