Python

Instalação

Python 3.9 ou mais novo. Não há nenhuma dependência obrigatória, e isso é deliberado: o exportador roda sobre urllib e gzip da biblioteca padrão, então instalar o mcpd não acrescenta nada ao lockfile e não pode conflitar com aquilo de que o seu agente já depende.

O extra httpx acrescenta reúso de conexão e mais nada. O SDK se comporta de forma idêntica sem ele.

Construir o cliente

Um cliente por processo. Todo argumento também lê uma variável de ambiente, então o mesmo código roda em desenvolvimento e em produção sem nenhuma edição. Mcpd() sem argumento nenhum fala com um daemon em 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",
)

O código vence o ambiente, e o ambiente vence o padrão.

enabled=False transforma toda chamada de registro em uma operação vazia que ainda devolve objetos válidos, então você pode desligar o mcpd nos testes sem mudar o código que está sendo testado. capture_payloads=False mantém prompts e respostas dentro do processo: os spans continuam sendo registrados com o seu tempo, o seu status, o seu uso, o seu modelo, a sua correlação e os seus erros, e input e output são descartados antes de chegar à fila. Um span pode dizer o contrário para si mesmo com capture_payloads=True nos seus campos de início, e no sentido inverso também.

Registrar uma execução

mcpd.trace(name) começa uma execução e mcpd.span(name) começa uma etapa abaixo do que estiver ativo no momento. Os dois são context managers: o bloco encerra o span na saída e registra a exceção, se o bloco levantou alguma, e depois deixa essa exceção passar sem alteração. O seu próprio tratamento de erros não muda por estar sendo 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

Sem um bloco with, chame run.end() você mesmo. De um jeito ou de outro o span é exportado quando começa, então um span que nunca termina fica visível como um span que nunca terminou, que costuma ser exatamente o que você está procurando.

# 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

O span ativo é uma contextvars.ContextVar. Ele acompanha um await para dentro de uma task e não vaza entre threads, então uma função auxiliar três quadros abaixo abre um filho do pai certo sem que nenhum chamador precise passar um argumento por todo o caminho.

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

O que sai disso é a árvore, com os tempos e os payloads anexados.

Entre serviços, o contexto viaja em um cabeçalho traceparent do W3C. Um cabeçalho malformado começa um trace novo em vez de falhar a requisição que o carregou. Uma decisão de não amostrar que chega de fora é obedecida, porque um trace que alguns serviços guardam e outros descartam não é um trace.

# 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 e erros

As contagens de tokens são o que o mcpd precifica na ingestão, então o custo no trace é o custo que foi cobrado de você. Informe o que o provedor relatou. set_error aceita uma string ou uma exceção, coloca o status em error no mesmo movimento, e sempre preenche uma mensagem: "alguma coisa falhou" não dá para agir em cima.

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

Os outros setters são set_input, set_output, set_metadata, set_attributes, set_tags, set_entity, set_cost, set_status e add_link. Todos devolvem o span, e qualquer um deles também pode ser passado como palavra-chave quando o span começa. external_parent_span_id= nomeia um pai de um sistema de tracing que não é o mcpd, carregado para correlação e nunca usado para definir a árvore.

Payloads, sinks e spans já concluídos

Um sink é para onde vai um lote drenado. O transporte HTTP é o padrão. Um RecordingSink guarda todo lote em memória, então um teste verifica o que a instrumentação registrou sem nenhum servidor; um ConsoleSink imprime uma linha JSON por sinal no stderr. Um sink que levanta exceção é um lote falho, contado e reportado, nunca uma exceção no seu código. stats() responde o que o cliente fez desde que foi construído.

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) enfileira um SpanRecord que já terminou, com os ids, o pai e os tempos como você os tem: o que um adaptador para outro sistema de tracing ou uma importação em lote já carrega. Ele nunca é reaninhado, nunca tem os tempos refeitos nem é reamostrado, e um registro malformado é descartado e contado em stats().rejected, nunca levantado como exceção.

Execuções duráveis

O mesmo cliente serve as execuções duráveis do daemon. workflow() constrói a definição e guarda os corpos das etapas, como chamadas ou como decoradores; um Worker a registra, faz long polling por leases em um pool de threads e serve cada um, e mcpd.runs, mcpd.events e mcpd.definitions espelham a superfície de execuções uma a uma. Um corpo levanta StepFailure com class_="retryable" para ser repetido dentro do orçamento da etapa, e registra spans abaixo da sua etapa através do StepContext. O que esses spans gastam é o uso da etapa: os tokens são somados conforme cada span termina e o último modelo visto vence, então o daemon precifica a etapa a partir do que o trace mostra; ctx.set_usage sobrescreve isso campo a campo. O Worker aceita quatro hooks além de on_error (on_lease, on_step_start, on_step_end, on_conflict), e worker.stats() conta leases tomados, corpos em andamento, concluídos, falhas por classe, conflitos por motivo e leases que não puderam ser servidos.

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

O que cada construção faz, o protocolo de lease e a API de execuções estão em Workflows, Workers e Execuções.

Notas e feedback

min e max são só por palavra-chave e não têm padrão. Um número solto, sem escala declarada, não pode ser comparado com nada, então a escala é exigida no ponto da chamada em vez de ser adivinhada depois. O feedback é o julgamento de uma pessoa e fica separado de uma nota: os dois respondem perguntas diferentes e nunca devem entrar na mesma média.

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

Os dois se ligam ao trace do span ativo quando você não nomeia nenhum. Os logs e as métricas fazem o mesmo, então um pico em um gráfico abre a execução que o causou.

Flush e encerramento

Registrar acrescenta a uma fila em memória e retorna; uma thread daemon faz o HTTP. flush() envia o que está esperando e devolve False se não conseguiu fazer isso dentro do seu tempo limite. shutdown() faz o flush, para o worker e libera o transporte. Os dois são idempotentes. Dentro de um event loop chame aflush() e ashutdown(), que fazem o mesmo trabalho sem bloquear o loop.

Em uma função serverless isso importa mais do que em qualquer outro lugar. O runtime congela o processo no instante em que o seu handler retorna, e a thread do worker nunca mais roda, então um lote que ainda estava na fila se perde.

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}

O cliente também registra um hook atexit, então um script comum que esquece de encerrar ainda recebe o seu último lote. Esse hook não roda em um os._exit() nem em um sinal fatal, então ele é uma rede de segurança e não um substituto.

Com o que você pode contar

• A instrumentação nunca levanta exceção. Toda chamada de registro é protegida; uma falha é logada no máximo uma vez por intervalo e engolida. A observabilidade nunca pode derrubar aquilo que ela observa. As chamadas de leitura são a exceção e levantam mesmo, porque uma consulta pela qual você está esperando é outra coisa.
• Registrar não bloqueia. A latência do seu agente nunca depende de o mcpd ser rápido, nem de estar no ar.
• A memória é limitada. São max_queue sinais, 8192 por padrão. Quando ela enche, os mais antigos são descartados, mcpd.dropped conta cada um e um aviso é logado por intervalo. Um buffer sem limite transforma uma queda de rede em uma morte por falta de memória.
• A decisão de amostragem é tomada uma vez, na raiz, a partir do id do trace, e herdada por todo filho. Um traceparent que chega de fora sempre vence.
• Um span é exportado quando começa e atualizado quando termina, então uma execução que quebra no meio do caminho continua visível.
• Os horários vêm de time.time_ns() e vão para a rede em UTC com nove dígitos. Horário local deixa um trace distribuído ilegível.

Ler de volta

Os métodos de leitura espelham a API HTTP e devolvem os tipos gerados. Eles são síncronos e levantam McpdError em caso de falha. from_ e to aceitam um datetime ou uma forma abreviada como -15m, -1h ou -7d. Em código assíncrono, rode-os fora do loop com 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 uma execução cobre os filtros por completo. O modelo de dados lista cada campo que esses tipos carregam.