Python

Instalación

Python 3.9 o posterior. No hay dependencias obligatorias, y es a propósito: el exportador funciona con urllib y gzip de la biblioteca estándar, así que instalar mcpd no añade nada a un lockfile y no puede entrar en conflicto con lo que tu agente ya usa.

El extra httpx añade reutilización de conexiones y nada más. El SDK se comporta igual sin él.

Construir el cliente

Un cliente por proceso. Cada argumento lee también una variable de entorno, así que el mismo código corre en desarrollo y en producción sin ninguna edición. Mcpd() sin argumentos habla con un daemon en 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",
)

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

enabled=False convierte cada llamada de registro en una operación vacía que sigue devolviendo objetos válidos, así que puedes desactivar mcpd en las pruebas sin cambiar el código bajo prueba. capture_payloads=False mantiene los prompts y las completions dentro del proceso: los spans se siguen registrando con 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 capture_payloads=True en sus campos de inicio, y también al revés.

Registrar una ejecución

mcpd.trace(name) inicia una ejecución y mcpd.span(name) inicia un paso bajo lo que esté activo en ese momento. Los dos son gestores de contexto: el bloque cierra el span al salir y registra la excepción si la hubo, y después deja pasar esa excepción sin cambios. Tu propio manejo de errores no cambia por estar observado.

# 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

Sin un bloque with, llama tú mismo a run.end(). En cualquiera de los dos casos el span se exporta al empezar, así que un span que nunca termina se ve como un span que nunca terminó, que suele ser justo el que buscas.

# Sync and async are the same object. Both end the span on the way out and
# record the exception if the block raised.
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()

Contexto

El span activo es un contextvars.ContextVar. Sigue a un await dentro de una tarea y no se filtra entre hilos, así que una función auxiliar tres marcos más abajo abre un hijo del padre correcto sin que ningún llamador le pase un argumento.

def draft(question: str) -> str:
    # Nothing was passed in. The span still lands under the run, because the
    # active span is a contextvars.ContextVar and this call is inside the block.
    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?"))

Lo que sale es el árbol, con los tiempos y los payloads adjuntos.

Entre servicios, el contexto viaja en una cabecera traceparent de W3C. Una cabecera mal formada inicia una traza nueva en vez de hacer fallar la solicitud que la traía. Se respeta la decisión entrante de no muestrear, porque una traza que unos servicios guardan y otros descartan no es una traza.

# Outbound: hand the active span to the next service.
requests.post(url, headers={"traceparent": mcpd.traceparent()}, json=payload)

# Inbound: continue that trace rather than starting a new one.
with mcpd.trace("handler", traceparent=request.headers.get("traceparent")) as run:
    ...

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. Informa lo que te dijo el proveedor. set_error acepta una cadena o una excepción, pone status en error en el mismo movimiento y siempre rellena un mensaje: «algo falló» no sirve para actuar.

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

# A failure you caught and decided about yourself.
generation.set_error("the provider timed out", kind="TimeoutError", retryable=True)

# A point in time inside the span, with no duration of its own.
generation.event("first_token")

Los demás setters son set_input, set_output, set_metadata, set_attributes, set_tags, set_entity, set_cost, set_status y add_link. Todos devuelven el span, y cualquiera de ellos se puede pasar también como argumento con nombre cuando el span empieza. external_parent_span_id= nombra un padre de un sistema de trazas que no es mcpd, que se lleva para correlacionar y nunca se usa para anidar.

Payloads, sinks y spans ya terminados

Un sink es adonde va un lote vaciado. El transporte HTTP es el valor por defecto. Un RecordingSink guarda cada lote en memoria, así que una prueba comprueba lo que registró la instrumentación sin servidor; un ConsoleSink imprime una línea JSON por señal en stderr. Un sink que lanza es un lote fallido, contado e informado, nunca una excepción en tu código. stats() responde qué ha hecho el cliente desde que se construyó.

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) encola un SpanRecord que ya terminó, con los ids, el padre y los tiempos tal como los tienes: lo que trae un adaptador de otro sistema de trazas o una importación por lotes. Nunca se le cambia el padre, ni los tiempos, ni el muestreo, y un registro mal formado se descarta y se cuenta en stats().rejected, nunca se lanza.

Ejecuciones duraderas

El mismo cliente sirve las ejecuciones duraderas del daemon. workflow() construye la definición y guarda los cuerpos de los pasos, como llamadas o como decoradores; un Worker la registra, hace long polling para obtener leases sobre un pool de hilos y atiende cada uno, y mcpd.runs, mcpd.events y mcpd.definitions reflejan la superficie de ejecuciones una a una. Un cuerpo lanza StepFailure con class_="retryable" para que se reintente dentro del presupuesto del paso, y registra spans bajo su paso a través del StepContext. Lo que gastan esos spans es el uso del paso: los tokens se suman según termina cada span y gana el último modelo visto, de modo que el daemon cobra el paso a partir de lo que muestra la traza; ctx.set_usage lo anula campo a campo. Worker acepta cuatro hooks junto a on_error (on_lease, on_step_start, on_step_end, on_conflict), y worker.stats() cuenta leases tomados, cuerpos en vuelo, completados, fallidos por clase, conflictos por motivo y leases que no se pueden atender.

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

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

Puntuaciones y feedback

min y max son solo por nombre y no tienen valor por defecto. Un número suelto sin escala declarada no se puede comparar con nada, así que la escala se exige en el punto de llamada en vez de adivinarse después. El feedback es el juicio de una persona y se mantiene aparte de una puntuación: los dos responden preguntas distintas y nunca se deben promediar juntos.

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

Los dos se adjuntan a la traza del span activo cuando no nombras ninguna. Los logs y las métricas hacen lo mismo, así que un pico en un gráfico abre la ejecución que lo causó.

Flush y apagado

Registrar añade a una cola en memoria y vuelve; un hilo daemon hace el HTTP. flush() envía lo que está esperando y devuelve False si no lo consiguió dentro de su tiempo límite. shutdown() hace flush, detiene el worker y libera el transporte. Los dos son idempotentes. Dentro de un bucle de eventos, llama a aflush() y ashutdown(), que hacen el mismo trabajo sin bloquear el bucle.

En una función serverless esto importa más que en ningún otro sitio. El runtime congela el proceso en cuanto tu handler devuelve, y el hilo del worker no vuelve a correr, así que se pierde el lote que seguía en la cola.

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

    # The batch is still in memory here. The runtime freezes this process the
    # moment the handler returns, and the worker thread never runs again.
    mcpd.flush()
    return {"statusCode": 200, "body": answer}

El cliente también registra un hook atexit, así que un script normal que se olvida de apagarlo aún recibe su último lote. Ese hook no se ejecuta con os._exit() ni con una señal fatal, así que es una red de seguridad y no un sustituto.

Con qué puedes contar

• La instrumentación nunca lanza. Cada llamada de registro está protegida; un fallo se anota como mucho una vez por intervalo y se traga. La observabilidad nunca debe tumbar aquello que observa. Las llamadas de lectura son la excepción y sí lanzan, porque una consulta que estás esperando es otra cosa.
• Registrar no bloquea. La latencia de tu agente nunca depende de que mcpd sea rápido, ni de que esté en pie.
• La memoria está acotada. max_queue señales, 8192 por defecto. Cuando está llena se descartan las más antiguas, mcpd.dropped las cuenta y se anota un aviso por intervalo. 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, a partir del id de traza, y la hereda cada hijo. Un traceparent entrante siempre gana.
• Un span se exporta cuando empieza y se actualiza cuando termina, así que una ejecución que se cae a mitad de vuelo sigue siendo visible.
• Las marcas de tiempo salen de time.time_ns() y viajan por el cable en UTC con nueve dígitos. La hora local hace ilegible una traza distribuida.

Volver a leerlo

Los métodos de lectura reflejan la API HTTP y devuelven los tipos generados. Son síncronos y lanzan McpdError cuando fallan. from_ y to aceptan un datetime o una forma abreviada como -15m, -1h o -7d. En código async, ejecútalos fuera del bucle 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}")

Encontrar una ejecución cubre los filtros por completo. El modelo de datos lista todos los campos que llevan estos tipos.