Python

Installazione

Python 3.9 o successivo. Non ci sono dipendenze obbligatorie, ed è voluto: l’exporter gira su urllib e gzip della libreria standard, quindi installare mcpd non aggiunge niente a un lockfile e non può entrare in conflitto con quello da cui il tuo agente già dipende.

L’extra httpx aggiunge il riuso delle connessioni e nient’altro. Senza, l’SDK si comporta esattamente allo stesso modo.

Costruire il client

Un client per processo. Ogni argomento legge anche una variabile d’ambiente, così lo stesso codice gira in sviluppo e in produzione senza una modifica. Mcpd() senza argomenti parla con un daemon su 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",
)

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

enabled=False rende ogni chiamata di registrazione un’operazione a vuoto che restituisce comunque oggetti validi, così puoi spegnere mcpd nei test senza cambiare il codice sotto test. capture_payloads=False tiene i prompt e le completion dentro il processo: gli span vengono registrati lo stesso, con 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. Uno span può dire il contrario per sé con capture_payloads=True fra i campi con cui parte, e viceversa.

Registrare un’esecuzione

mcpd.trace(name) avvia un’esecuzione e mcpd.span(name) avvia un passo sotto quello che è attivo in quel momento. Tutti e due sono context manager: il blocco chiude lo span all’uscita e registra l’eccezione se ne è stata sollevata una, poi lascia passare quell’eccezione identica. La tua gestione degli errori non cambia per il fatto di essere osservata.

# 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

Senza un blocco with, chiama tu run.end(). In tutti e due i casi lo span viene esportato quando parte, quindi uno span che non finisce mai si vede come uno span che non è mai finito, che di solito è proprio quello che stai cercando.

# Il sincrono e l'asincrono sono lo stesso oggetto. Tutti e due chiudono lo
# span all'uscita e registrano l'eccezione se il blocco ne ha sollevata una.
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()

Contesto

Lo span attivo è un contextvars.ContextVar. Segue un await dentro un task e non trapela fra un thread e l’altro, così una funzione di appoggio tre livelli più in basso apre un figlio del padre giusto senza che nessun chiamante le passi un argomento.

def draft(question: str) -> str:
    # Non è stato passato niente. Lo span atterra comunque sotto l'esecuzione:
    # lo span attivo è un contextvars.ContextVar e la chiamata sta nel blocco.
    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?"))

Quello che ne esce è l’albero, con i tempi e i payload attaccati.

Fra due servizi il contesto viaggia in un header traceparent W3C. Un header malformato fa partire una traccia nuova anziché far fallire la richiesta che lo portava. Una decisione di non campionare arrivata da fuori viene rispettata, perché una traccia che alcuni servizi tengono e altri buttano non è una traccia.

# In uscita: passa lo span attivo al servizio successivo.
requests.post(url, headers={"traceparent": mcpd.traceparent()}, json=payload)

# In entrata: continua quella traccia invece di aprirne una nuova.
with mcpd.trace("handler", traceparent=request.headers.get("traceparent")) as run:
    ...

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. Riferisci quello che ti ha detto il provider. set_error accetta una stringa o un’eccezione, porta status a error nello stesso gesto, e riempie sempre un messaggio: «qualcosa è fallito» non dice a nessuno che cosa fare.

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

# Un errore che hai intercettato e su cui hai deciso tu.
generation.set_error("the provider timed out", kind="TimeoutError", retryable=True)

# Un istante dentro lo span, senza durata propria.
generation.event("first_token")

Gli altri setter sono set_input, set_output, set_metadata, set_attributes, set_tags, set_entity, set_cost, set_status e add_link. Tutti restituiscono lo span, e ognuno di loro si può anche passare come argomento con nome quando lo span parte. external_parent_span_id= nomina un padre che viene da un sistema di tracing che non è mcpd, portato per la correlazione e mai usato per costruire l’albero.

Payload, sink e span già conclusi

Un sink è il posto dove finisce un batch svuotato. Il trasporto HTTP è quello predefinito. Un RecordingSink tiene in memoria ogni batch, così un test verifica che cosa ha registrato la strumentazione senza un server; un ConsoleSink stampa una riga JSON per segnale su stderr. Un sink che solleva un’eccezione è un batch fallito, contato e riferito, mai un’eccezione nel tuo codice. stats() risponde che cosa ha fatto il client da quando è stato costruito.

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) mette in coda uno SpanRecord già concluso, con gli identificatori, il padre e i tempi come li hai tu: quello che ha sottomano un adattatore per un altro sistema di tracing o un’importazione in blocco. Non gli viene mai cambiato il padre, né i tempi, né il campionamento, e un record malformato viene scartato e contato in stats().rejected, mai sollevato come eccezione.

Esecuzioni durevoli

Lo stesso client serve le esecuzioni durevoli del daemon. workflow() costruisce la definizione e conserva i corpi dei passi, come chiamate o come decoratori; un Worker la registra, fa long polling per ottenere i lease su un pool di thread e serve ognuno di essi, e mcpd.runs, mcpd.events e mcpd.definitions rispecchiano uno a uno la superficie delle esecuzioni. Un corpo solleva StepFailure con class_="retryable" per essere ritentato entro il budget del suo passo, e registra span sotto il proprio passo attraverso lo StepContext. Quello che spendono quegli span è il consumo del passo: i token vengono sommati man mano che ogni span finisce e vince l’ultimo modello visto, così il daemon valorizza il passo su quello che mostra la traccia; ctx.set_usage lo sovrascrive campo per campo. Worker prende quattro hook accanto a on_error (on_lease, on_step_start, on_step_end, on_conflict), e worker.stats() conta i lease presi in carico, i corpi in volo, quelli completati, quelli falliti per classe, i conflitti per motivo e i lease che non si sono potuti servire.

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

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

Punteggi e feedback

min e max si passano solo con il nome e non hanno un valore predefinito. Un numero nudo, senza una scala dichiarata, non si può confrontare con niente, quindi la scala si impone nel punto della chiamata anziché indovinarla dopo. Il feedback è il giudizio di una persona e resta separato da un punteggio: i due rispondono a domande diverse e non vanno mai mediati insieme.

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

Tutti e due si attaccano alla traccia dello span attivo quando non ne nomini una. I log e le metriche fanno lo stesso, così un picco su un grafico apre l’esecuzione che lo ha causato.

Flush e arresto

Registrare aggiunge a una coda in memoria e ritorna; l’HTTP lo fa un thread daemon. flush() manda quello che è in attesa e restituisce False se non è riuscito a farlo entro il suo timeout. shutdown() fa il flush, ferma il worker e rilascia il trasporto. Tutti e due sono idempotenti. Dentro un event loop chiama aflush() e ashutdown(), che fanno lo stesso lavoro senza bloccare il loop.

In una funzione serverless questo conta più che altrove. Il runtime congela il processo nel momento in cui il tuo handler ritorna, e il thread del worker non gira mai più, quindi un batch ancora in coda è perso.

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

    # Qui il batch è ancora in memoria. Il runtime congela il processo nel
    # momento in cui l'handler ritorna, e il thread del worker non gira mai più.
    mcpd.flush()
    return {"statusCode": 200, "body": answer}

Un client registra anche un hook atexit, così uno script normale che si dimentica di arrestarlo manda comunque il suo ultimo batch. Quell’hook non gira su os._exit() né su un segnale fatale, quindi è una rete di sicurezza e non un sostituto.

Su che cosa puoi contare

• La strumentazione non solleva mai eccezioni. Ogni chiamata di registrazione è protetta; un guasto finisce in un log al massimo una volta per intervallo e poi viene inghiottito. L’osservabilità non deve mai far cadere la cosa che osserva. Le chiamate in lettura sono l’eccezione e sollevano davvero, perché una query che stai aspettando è un’altra cosa.
• Registrare non blocca. La latenza del tuo agente non dipende mai da quanto mcpd sia veloce, né dal fatto che sia attivo.
• La memoria è limitata. max_queue segnali, 8192 come impostazione predefinita. Quando è piena i più vecchi vengono scartati, mcpd.dropped li conta e viene scritto un avvertimento per intervallo. 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, a partire dall’id di traccia, e la ereditano tutti i figli. Un traceparent arrivato da fuori vince sempre.
• Uno span viene esportato quando parte e aggiornato quando finisce, così un’esecuzione che va in crash per strada resta visibile.
• I timestamp vengono da time.time_ns() e viaggiano sulla rete in UTC con nove cifre. L’ora locale rende illeggibile una traccia distribuita.

Rileggerla

I metodi di lettura rispecchiano l’API HTTP e restituiscono i tipi generati. Sono sincroni e sollevano McpdError quando falliscono. from_ e to accettano un datetime o una forma breve come -15m, -1h o -7d. Nel codice asincrono falli girare fuori dal loop con 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}")

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