Python

Installation

Python 3.9 ou plus récent. Il n’y a aucune dépendance obligatoire, et c’est délibéré : l’exporter tourne sur urllib et gzip de la bibliothèque standard, si bien qu’installer mcpd n’ajoute rien à un fichier de verrouillage et ne peut entrer en conflit avec ce dont votre agent dépend déjà.

L’extra httpx ajoute la réutilisation des connexions et rien d’autre. Le SDK se comporte à l’identique sans lui.

Construire le client

Un client par processus. Chaque argument lit aussi une variable d’environnement, si bien que le même code tourne en développement et en production sans la moindre modification. Mcpd() sans argument s’adresse à un daemon sur http://localhost:4317.

from mcpd import FeedbackKind, Mcpd, SpanKind

# Everything here has an MCPD_* environment variable behind it, so the same code
# runs in development and in production with no edit. `Mcpd()` with no arguments
# talks to a local daemon.
mcpd = Mcpd(
    endpoint=os.environ.get("MCPD_ENDPOINT", "http://localhost:4317"),
    project="support",
    environment="development",
    service_name="support-agent",
    service_version="1.4.2",
)

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

enabled=False fait de chaque appel d’enregistrement une opération sans effet qui renvoie quand même des objets valides : vous pouvez donc désactiver mcpd dans les tests sans modifier le code testé. capture_payloads=False garde les prompts et les complétions dans le processus : les spans sont toujours enregistrés avec 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 capture_payloads=True dans ses champs de départ, et réciproquement.

Enregistrer une exécution

mcpd.trace(name) démarre une exécution et mcpd.span(name) démarre une étape sous ce qui est actif à ce moment-là. Les deux sont des gestionnaires de contexte : le bloc termine le span en sortant et enregistre l’exception s’il en a levé une, puis laisse passer cette exception telle quelle. Votre propre gestion des erreurs n’est pas altérée par le fait d’être observée.

# The root span is the run. Everything opened inside it becomes a child,
# because the active span lives in a `contextvars.ContextVar`.
with mcpd.trace("support_agent", session_id="sess-8121", user_id=user_id,
                tags=["support", "orders"]) as run:
    run.set_input({"question": question})

    # A model call. `set_usage` is what mcpd prices the run from, so the
    # cost on the trace is the cost you were actually billed.
    with mcpd.span("plan", kind=SpanKind.MODEL_GENERATION) as generation:
        generation.set_model("gpt-4o-mini", provider="openai")
        generation.set_input([
            {"role": "system", "content": "You are a support agent."},
            {"role": "user", "content": question},
        ])
        time.sleep(0.12)  # stand-in for the provider round trip
        generation.set_output("I should look the order up first.")
        generation.set_usage(input_tokens=812, output_tokens=96, cache_read=128)

    # A tool call that fails. The `with` block records the exception on the
    # span, marks it as an error, and lets the exception through unchanged.
    order = None
    try:
        with mcpd.span("lookup_order", kind=SpanKind.TOOL_CALL) as tool:
            tool.set_entity("tool", entity_id="orders.lookup", name="Order lookup")
            tool.set_input({"order_id": "A-7741"})
            order = lookup_order("A-7741")
            tool.set_output(order)
    except TimeoutError as exc:
        # The span already carries the error. This is the agent deciding
        # what to do about it, which is a different question.
        mcpd.log("warn", "order lookup failed, answering from context", {"error": str(exc)})
        mcpd.metric("support.tool_failures", 1, "counter", labels={"tool": "orders.lookup"})

    reply = ("I could not reach the order system just now. Your order A-7741 "
             "shipped on Tuesday and tracking should update within the hour.")
    run.set_output(reply)
    run.set_metadata({"used_fallback": order is None})
    return reply, run.trace_id

Sans bloc with, appelez run.end() vous-même. Le span est exporté dès qu’il démarre dans les deux cas : un span qui ne se termine jamais est donc visible comme un span qui ne s’est jamais terminé, et c’est en général celui que vous cherchez.

# Le synchrone et l’asynchrone sont le même objet. Les deux terminent le span
# en sortant et enregistrent l’exception si le bloc en a levé une.
with mcpd.trace("nightly_summariser") as run:
    with mcpd.span("fetch", kind=SpanKind.TOOL_CALL) as step:
        step.set_output({"documents": 42})

async with mcpd.trace("agent") as run:
    async with mcpd.span("generate", kind=SpanKind.MODEL_GENERATION) as generation:
        generation.set_output(await model.complete(prompt))
    await mcpd.aflush()

Contexte

Le span actif est un contextvars.ContextVar. Il suit un await jusque dans une tâche et ne fuit pas entre threads : une fonction utilitaire trois cadres plus bas ouvre donc un enfant du bon parent sans qu’aucun appelant ait à lui faire passer un argument.

def draft(question: str) -> str:
    # Rien n’a été passé en argument. Le span se place quand même sous
    # l’exécution, car le span actif est un contextvars.ContextVar et cet
    # appel est à l’intérieur du bloc.
    with mcpd.span("draft", kind=SpanKind.MODEL_GENERATION) as generation:
        generation.set_input(question)
        return "..."


with mcpd.trace("support_agent") as run:
    run.set_output(draft("where is my order?"))

Ce qui en ressort est l’arbre, avec les temps et les payloads attachés.

Entre services, le contexte voyage dans un en-tête W3C traceparent. Un en-tête mal formé démarre une nouvelle trace plutôt que de faire échouer la requête qui le transportait. Une décision entrante de ne pas échantillonner est respectée, car une trace que certains services gardent et que d’autres jettent n’est pas une trace.

# En sortie, passez le span actif au service suivant.
requests.post(url, headers={"traceparent": mcpd.traceparent()}, json=payload)

# En entrée, continuez cette trace plutôt que d’en démarrer une nouvelle.
with mcpd.trace("handler", traceparent=request.headers.get("traceparent")) as run:
    ...

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é. Reportez ce que le fournisseur vous a indiqué. set_error accepte une chaîne ou une exception, place status sur error du même geste, et renseigne toujours un message : « quelque chose a échoué » ne dit pas quoi faire.

generation.set_model("gpt-4o-mini", provider="openai")
generation.set_usage(input_tokens=812, output_tokens=96, cache_read=128)

# Un échec que vous avez intercepté et sur lequel vous avez tranché vous-même.
generation.set_error("the provider timed out", kind="TimeoutError", retryable=True)

# Un instant à l’intérieur du span, sans durée propre.
generation.event("first_token")

Les autres setters sont set_input, set_output, set_metadata, set_attributes, set_tags, set_entity, set_cost, set_status et add_link. Tous renvoient le span, et chacun peut aussi être passé en argument nommé au démarrage du span. external_parent_span_id= nomme un parent venu d’un système de tracing autre que mcpd, porté pour la corrélation et jamais utilisé pour la filiation.

Payloads, sinks et spans terminés

Un sink est l’endroit où part un lot vidé. Le transport HTTP est le sink par défaut. Un RecordingSink garde chaque lot en mémoire : un test vérifie ainsi ce que l’instrumentation a enregistré sans serveur. Un ConsoleSink imprime une ligne JSON par signal sur stderr. Un sink qui lève une exception donne un lot en échec, compté et signalé, jamais une exception dans votre code. stats() répond ce que le client a fait depuis sa construction.

from mcpd import Mcpd, RecordingSink

sink = RecordingSink()
mcpd = Mcpd(sink=sink)
with mcpd.trace("answer") as run:
    run.set_usage(input_tokens=12)
mcpd.flush()
assert sink.spans[0].usage.input_tokens == 12
print(mcpd.stats())  # queued, dropped, capacity, exported, failed, rejected

mcpd.record(span) met en file un SpanRecord déjà terminé, avec les identifiants, le parent et les temps tels que vous les détenez : ce dont dispose un adaptateur pour un autre système de tracing ou un import par lots. Il n’est jamais re-parenté, re-daté ni ré-échantillonné, et un enregistrement mal formé est écarté et compté dans stats().rejected, jamais levé.

Exécutions durables

Le même client dessert les exécutions durables du daemon. workflow() construit la définition et retient le corps des étapes, sous forme d’appels ou de décorateurs ; un Worker l’enregistre, attend des leases en long polling sur un pool de threads et sert chacun d’eux, et mcpd.runs, mcpd.events et mcpd.definitions reprennent un pour un la surface des exécutions. Un corps d’étape lève StepFailure avec class_="retryable" pour être réessayé dans le budget de l’étape, et enregistre des spans sous son étape via le StepContext. Ce que dépensent ces spans est la consommation de l’étape : les tokens sont additionnés à la fin de chaque span et le dernier modèle vu l’emporte, si bien que le daemon tarife l’étape d’après ce que montre la trace ; ctx.set_usage la remplace champ par champ. Worker accepte quatre hooks à côté de on_error (on_lease, on_step_start, on_step_end, on_conflict), et worker.stats() compte les leases prises, les corps en cours, les achèvements, les échecs par classe, les conflits par motif et les leases non servables.

support = workflow("support_ticket")


@support.step("classify")
def classify(ctx, input, prior):
    return {"kind": "question" if input["ticket"].startswith("Q") else "bug"}


@support.agent("draft", input={"kind": "${steps.classify.output.kind}"})
def draft(ctx, input, prior):
    # Your own model call, recorded under this step in the run's trace. What
    # the span spends is the step's usage: the daemon prices it from here.
    with ctx.span("model_call", kind="model_generation") as span:
        span.set_model("claude-sonnet-4", "anthropic")
        span.set_usage(input_tokens=812, output_tokens=37)
        text = f"an answer to a {input['kind']}"
    return text


support.wait_for("approval", event="approved", key="${input.ticket}",
                 timeout_ms=7 * 24 * 3_600_000,
                 when=when("steps.classify.output.kind", "eq", "bug"))


@support.step("send", input={"text": "${steps.draft.output}"}, effect="at_most_once",
              retry=RetryPolicy(max_attempts=3, backoff="exponential", initial_ms=500))
def send(ctx, input, prior):
    if not deliver(input["text"], ctx.idempotency_key):
        raise StepFailure("mail relay busy", class_="retryable", retry_after_ms=2_000)
    return {"sent": True}


support.output({"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

min et max se passent uniquement par mot-clé et n’ont pas de valeur par défaut. Un nombre nu, sur une échelle non déclarée, ne peut être comparé à rien : l’échelle est donc exigée à l’appel plutôt que devinée plus tard. Le feedback est le jugement d’une personne et reste distinct d’un score : les deux répondent à des questions différentes et ne doivent jamais être moyennés ensemble.

# A machine evaluation. `min` and `max` are required: a bare number with no
# declared scale cannot be compared to anything.
mcpd.score(
    "answer_relevance",
    value=round(random.uniform(0.6, 0.95), 2),
    min=0.0,
    max=1.0,
    trace_id=trace_id,
    reason="named the order and gave a delivery window, without the live lookup",
)

# A human evaluation. Kept separate from a score, because a person's
# judgement and a model's must never be averaged together.
mcpd.feedback(
    FeedbackKind.THUMBS,
    "up",
    trace_id=trace_id,
    comment="fine given the outage",
    author_kind="end_user",
)

Les deux se rattachent à la trace du span actif quand vous n’en nommez aucune. Les logs et les métriques font de même, si bien qu’un pic sur un graphique ouvre l’exécution qui l’a causé.

Flush et arrêt

Enregistrer ajoute à une file d’attente en mémoire et rend la main ; un thread daemon se charge du HTTP. flush() envoie ce qui attend et renvoie False s’il n’y est pas parvenu dans son délai. shutdown() vide la file, arrête le worker et libère le transport. Les deux sont idempotents. À l’intérieur d’une boucle d’événements, appelez aflush() et ashutdown(), qui font le même travail sans bloquer la boucle.

Dans une fonction serverless, cela compte plus que partout ailleurs. Le runtime gèle le processus dès que votre handler rend la main, et le thread worker ne tourne plus jamais : un lot encore en file est donc perdu.

def handler(event, context):
    with mcpd.trace("answer") as run:
        run.set_input(event["body"])
        answer = agent(event["body"])
        run.set_output(answer)

    # Le lot est encore en mémoire ici. Le runtime gèle ce processus dès que le
    # handler rend la main, et le thread worker ne tourne plus jamais.
    mcpd.flush()
    return {"statusCode": 200, "body": answer}

Un client enregistre aussi un hook atexit, si bien qu’un script ordinaire qui oublie de s’arrêter proprement obtient quand même son dernier lot. Ce hook ne s’exécute pas sur os._exit() ni sur un signal fatal : c’est un filet de sécurité, pas un remplacement.

Ce sur quoi vous pouvez compter

• L’instrumentation ne lève jamais d’exception. Chaque appel d’enregistrement est protégé ; un échec est journalisé au plus une fois par intervalle puis avalé. L’observabilité ne doit jamais faire tomber ce qu’elle observe. Les appels de lecture font exception et lèvent bel et bien une erreur, car une requête que vous attendez est autre chose.
• L’enregistrement ne bloque pas. La latence de votre agent ne dépend jamais de la vitesse de mcpd, ni de sa disponibilité.
• La mémoire est bornée. max_queue signaux, 8192 par défaut. Quand elle est pleine, les plus anciens sont écartés, mcpd.dropped les compte et un avertissement est journalisé par intervalle. 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, à partir de l’identifiant de trace, puis héritée par chaque enfant. Un traceparent entrant l’emporte toujours.
• Un span est exporté quand il démarre et mis à jour quand il se termine, si bien qu’une exécution qui plante en plein vol reste visible.
• Les horodatages viennent de time.time_ns() et partent sur le réseau en UTC avec neuf chiffres. L’heure locale rend une trace distribuée illisible.

Relire ce qui a été enregistré

Les méthodes de lecture reflètent l’API HTTP et renvoient les types générés. Elles sont synchrones et lèvent McpdError en cas d’échec. from_ et to acceptent un datetime ou un raccourci tel que -15m, -1h ou -7d. En code asynchrone, exécutez-les hors de la boucle avec asyncio.to_thread.

# Nothing has been sent yet: recording appends to an in-memory queue and a
# background worker does the I/O. Flush before you read your own writes.
mcpd.flush()

detail = mcpd.get_trace(trace_id, include=["logs", "scores", "feedback"])
print(f"\n{detail.trace.name}  {detail.trace.duration_ns / 1e6:.0f} ms  "
      f"{detail.trace.span_count} spans  {detail.trace.error_count} errors")
for span in detail.spans:
    cost = f"  ${span.cost.total_cost:.6f}" if span.cost else ""
    print(f"  {span.kind.value:<18} {span.name:<14} {span.status.value}{cost}")

summary = mcpd.analytics_summary("-1h", "-0s")
print(f"\nlast hour  traces={summary.trace_count}  errors={summary.error_count}  "
      f"p95={summary.latency.p95:.0f}ms  cost=${summary.total_cost:.4f}")

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