Elixir

Instalação

Acrescente ao mix.exs. Ele não traz nada junto.

defp deps do
  [{:mcpd, "~> 0.1"}]
end

Essa é a lista de dependências inteira, e isso é deliberado. O OTP 27 já traz um codec JSON (:json) e um cliente HTTP (:inets), então uma biblioteca cujo trabalho é observar o seu sistema não precisa virar um motivo para o build do seu sistema mudar. Os outros cinco SDKs fazem a mesma promessa nas suas próprias linguagens.

Construir o cliente

Toda opção cai de volta na sua variável de ambiente, e o código vence. Mcpd.start_link() sem argumento nenhum fala com um daemon local.

# Every option also reads an environment variable (MCPD_ENDPOINT, MCPD_PROJECT,
# MCPD_API_KEY, ...), so `Mcpd.start_link()` on its own works too.
{:ok, mcpd} =
  Mcpd.start_link(
    endpoint: "http://localhost:4317",
    project: "support-desk",
    environment: "development",
    service_name: "support-agent",
    service_version: "1.4.0"
  )

Coloque-o na sua árvore de supervisão e ele se comporta como qualquer outro worker:

children = [
  {Mcpd, project: "support-desk", environment: "production"}
]

Supervisor.start_link(children, strategy: :one_for_one)

Registrar uma execução

span é uma macro, então uma execução é um bloco e o valor dela é o valor do bloco.

question = "where is my order 41?"

# The block's value is the value of `span`. Inside it, this span is the active
# one for the process, so the helpers below find their parent without being
# handed it.
trace_id =
  Mcpd.span mcpd, "support_agent",
    kind: :agent_run,
    input: question,
    session_id: "session-8f21",
    user_id: "customer-4477",
    tags: ["support", "email"] do
    Mcpd.log(mcpd, :info, "handling a question", %{"channel" => "email"})

    plan = plan_answer.(mcpd, question)
    order = look_up_order.(mcpd, 41)
    estimate = estimate_delivery.(mcpd, 41)

    _answer =
      [plan, order, estimate && "It should arrive #{estimate}."]
      |> Enum.reject(&is_nil/1)
      |> Enum.join(" ")

    Mcpd.metric(mcpd, "support.answers", 1, kind: :counter, unit: "{answer}")
    Mcpd.Context.current().trace_id
  end

Macros precisam ser requeridas antes de poderem ser chamadas, o que é uma regra da linguagem e não uma escolha daqui. use Mcpd faz isso por você:

defmodule MyAgent do
  use Mcpd

  def answer(mcpd, question) do
    Mcpd.span mcpd, "agent", kind: :agent_run, input: question do
      do_the_work(question)
    end
  end
end

Se o corpo já for uma função em vez de um bloco, chame Mcpd.with_span/4. É a mesma coisa, sem a macro.

A árvore se monta sozinha

O span ativo vive no dicionário do processo, e esse é o lugar certo em Elixir especificamente, não um atalho. Aqui o processo é a unidade de concorrência, então duas requisições atendidas ao mesmo tempo não conseguem ver o span uma da outra, e nada precisa ser passado por todas as funções para manter a árvore inteira.

Um filho aberto três chamadas abaixo encontra o seu pai sozinho:

Mcpd.span mcpd, "agent", kind: :agent_run do
  plan(mcpd)          # opens a child of "agent" without being told
end

defp plan(mcpd) do
  Mcpd.span mcpd, "plan_answer", kind: :model_generation, model: "gpt-4o" do
    call_the_model()
  end
end

A consequência que vale conhecer: um span aberto dentro de Task.async começa um trace novo, porque aquilo é outro processo. Passe o trace adiante com Mcpd.traceparent/0 e Mcpd.continue/2, o mesmo par que você usaria através de uma fronteira HTTP.

Nada bloqueia, e nada levanta exceção

Registrar um span envia uma mensagem ao worker e retorna. A rede acontece naquele worker, então a latência do seu agente nunca depende de o mcpd ser rápido, nem de estar no ar, nem de estar rodando.

Uma exceção levantada dentro de um bloco é registrada no span como um erro e depois levantada de novo sem alteração. O trabalho do mcpd é ver o que aconteceu, nunca decidir o que o seu código faz a respeito.

A fila tem teto, e quando ela enche os sinais mais antigos são descartados. Essa direção é escolhida: quando o daemon está inacessível, os sinais mais novos descrevem o que está acontecendo agora, e são esses que alguém vai querer quando ele voltar. Mcpd.stats/1 responde o que o cliente fez: queued, dropped, capacity, exported, failed e rejected.

log_level:, ou MCPD_LOG_LEVEL, define um piso abaixo do Mcpd.log/4. O padrão é debug, que guarda tudo. Uma linha abaixo do piso é descartada antes de o registro ser construído, então uma linha de debug em uma implantação de produção custa uma comparação. O piso vale só para os logs; spans, métricas, notas e feedback não são afetados.

Mantenha os payloads dentro do processo

capture_payloads: false, ou MCPD_CAPTURE_PAYLOADS=false, mantém prompts e respostas onde eles estão. Os spans continuam saindo com o seu tempo, o seu status, o seu uso, o seu modelo, a sua correlação e os seus erros; input:, output: e o valor capturado de um bloco são descartados antes de chegar à fila, então nada sensível vai para a fila e nada sensível sai em um flush. Um span sozinho pode dizer o contrário para si mesmo com capture_payloads: true.

Verifique o que você registrou

Um sink é para onde vai um lote drenado. O transporte HTTP é o padrão; Mcpd.Sink.Recorder guarda todo lote em memória, então um teste lê de volta os spans, logs, métricas, notas e feedback que a instrumentação registrou sem nenhum daemon, e Mcpd.Sink.Console imprime uma linha JSON por sinal no stderr. Um sink seu implementa Mcpd.Sink; um que levanta exceção é um lote falho, contado e reportado, nunca levantado para quem chamou.

Mcpd.record/3 enfileira um span que já terminou, como veio: com os seus próprios ids, o seu pai, o seu início e o seu fim, nunca reaninhado, nunca com os tempos refeitos nem reamostrado. É isso que um adaptador para outro sistema de tracing ou uma importação em lote tem em mãos. Um registro cujos ids não tenham o formato do OpenTelemetry é recusado e contado, e nunca levanta exceção. external_parent_span_id: em qualquer span carrega um pai de um sistema de tracing que não é o mcpd, só para correlação.

Execuções duráveis

O mesmo cliente serve as execuções duráveis do daemon. Mcpd.Workflow constrói a definição e guarda os corpos das etapas como funções de aridade três; um Mcpd.StepWorker a registra, faz long polling por leases e serve cada um em um processo próprio, e Mcpd.start_run/3, Mcpd.watch_run/3, Mcpd.publish_event/4 e os demais espelham a superfície de execuções uma a uma. Um corpo levanta Mcpd.StepException com class: :retryable para ser repetido dentro do orçamento da etapa, e registra spans abaixo da sua etapa através de Mcpd.StepContext.span/4. Todo span que o corpo fecha soma o seu uso ao da etapa, e o último modelo visto vence, então o daemon precifica o que o trace mostra; set_usage/2 é a sobrescrita explícita. on_lease, on_step_start, on_step_end e on_conflict ficam ao lado de on_error no worker, e StepWorker.stats/1 conta o que ele tomou, concluiu, falhou por classe e abandonou por motivo.

support =
  Workflow.new("support_ticket")
  |> Workflow.step("classify", fn _ctx, input, _prior ->
    kind = if String.starts_with?(input["ticket"], "Q"), do: "question", else: "bug"
    %{"kind" => kind}
  end)
  |> Workflow.agent(
    "draft",
    fn ctx, input, _prior ->
      # Your own model call, recorded under this step in the run's trace. Its
      # usage is the step's: the daemon prices what the trace shows.
      opts = [kind: :model_generation, model: "claude-sonnet-4", provider: "anthropic",
        usage: %{"input_tokens" => 812, "output_tokens" => 37}]
      StepContext.span(ctx, "model_call", opts, fn -> :ok end)

      "an answer to a #{input["kind"]}"
    end,
    input: %{"kind" => "${steps.classify.output.kind}"}
  )
  |> Workflow.wait_for("approval",
    event: "approved",
    key: "${input.ticket}",
    timeout_ms: 7 * 24 * 3_600_000,
    when: where("steps.classify.output.kind", :eq, "bug")
  )
  |> Workflow.step(
    "send",
    fn ctx, input, _prior ->
      unless deliver.(input["text"], ctx.idempotency_key) do
        raise Mcpd.StepException,
          message: "mail relay busy", class: :retryable, retry_after_ms: 2_000
      end

      %{"sent" => true}
    end,
    input: %{"text" => "${steps.draft.output}"},
    effect: :at_most_once,
    retry: [max_attempts: 3, backoff: :exponential, initial_ms: 500]
  )
  |> Workflow.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

# A machine evaluation. `min` and `max` are required: 7 is excellent out of 10
# and poor out of 100, and a score with no scale cannot be compared.
Mcpd.score(mcpd, trace_id, "answer_relevance", 0.86,
  min: 0,
  max: 1,
  reason: "answered the question asked, without the delivery estimate"
)

# A person's judgement, from the thumbs-up control in your own product.
Mcpd.feedback(mcpd, trace_id, :thumbs, "up",
  comment: "fast and clear",
  author_id: "customer-4477"
)

# Nothing above touched the network. This is where it goes.
:ok = Mcpd.flush(mcpd)

Ler de volta

{:ok, detail} = Mcpd.get_trace(mcpd, trace_id, include: ["scores", "feedback"])

IO.puts("#{detail.trace.name}: #{length(detail.spans)} spans")

for span <- detail.spans do
  ms = div(span.duration_ns || 0, 1_000_000)
  error = (span.error && span.error.message) || ""
  IO.puts("  #{span.kind} #{span.name} #{ms}ms #{error}")
end

Processos de vida curta precisam fazer flush

Um script ou uma task que termina rápido perde tudo que ainda não tinha sido enviado. Mcpd.shutdown/1 faz o flush e para o worker, e pode ser chamado duas vezes sem problema.

Depois disso

• Todo comportamento que este SDK divide com os outros cinco: o contrato dos SDKs.
• O que um span carrega: Modelo de dados.
• Como encontrar depois a execução que interessa a você: Encontrar uma execução.