Elixir

インストール

mix.exs に追加します。ほかには何も付いてきません。

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

依存の一覧はこれで全部で、そうしているのは意図的です。OTP 27 には JSON コーデック（:json）と HTTP クライアント（:inets）が 入っています。システムを見張るのが仕事のライブラリが、そのシステムのビルドを 変える理由になってはいけません。ほかの 5 つの SDK も、それぞれの言語で同じ 約束をしています。

クライアントの構築

どのオプションも、指定がなければ環境変数を読みます。コードのほうが優先されます。 引数なしの Mcpd.start_link() は、ローカルのデーモンに接続します。

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

スーパービジョンツリーに入れれば、ほかのワーカーと同じように振る舞います。

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

Supervisor.start_link(children, strategy: :one_for_one)

実行の記録

span はマクロなので、実行はブロックとして書け、その値はブロックの 値になります。

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

マクロは、呼ぶ前に require しておく必要があります。これはこの SDK の選択では なく、言語のルールです。use Mcpd がそれを済ませます。

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

本体がブロックではなく、すでに関数になっている場合は、代わりに Mcpd.with_span/4 を呼びます。マクロを使わないだけで、中身は 同じです。

ツリーの組み上がり方

アクティブなスパンはプロセス辞書に入ります。これは手抜きではなく、Elixir に 限っては正しい置き場所です。ここでは並行性の単位がプロセスなので、同時に 処理される 2 つのリクエストが互いのスパンを見ることはありませんし、ツリーを つなぐために何かをすべての関数へ引き回す必要もありません。

3 段下の呼び出しで開いた子も、自分で親を見つけます。

Mcpd.span mcpd, "agent", kind: :agent_run do
  plan(mcpd)          # 言われなくても "agent" の子を開く
end

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

知っておくべき帰結があります。Task.async の中で開いたスパンは、 別のプロセスなので新しいトレースを始めます。トレースを渡すには Mcpd.traceparent/0 と Mcpd.continue/2 を使います。 HTTP の境界をまたぐときと同じ組み合わせです。

ブロックしない、例外を投げない

スパンの記録は、ワーカーにメッセージを送って戻るだけです。ネットワークはその ワーカーが担うので、エージェントのレイテンシが、mcpd の速さや、mcpd が動いて いるかどうか、そもそも起動しているかどうかに左右されることはありません。

ブロックの中で送出された例外は、スパンにエラーとして記録され、そのまま再送出 されます。mcpd の仕事は起きたことを見ることであって、それに対してあなたの コードがどうするかを決めることではありません。

キューには上限があり、いっぱいになると古いシグナルから捨てられます。この向きは 選んだ結果です。デーモンに到達できないとき、いま起きていることを説明するのは 新しいシグナルのほうで、復旧したときに誰かが見たいのもそちらだからです。 Mcpd.stats/1 は、クライアントが何をしたかを返します。 queued、dropped、capacity、 exported、failed、rejected です。

log_level:、または MCPD_LOG_LEVEL は、 Mcpd.log/4 の下限を決めます。既定は debug で、 すべてを残します。下限より下の行はレコードを組み立てる前に捨てられるので、 本番環境のデバッグ行にかかるのは比較 1 回分です。この下限が効くのはログだけで、 スパン、メトリクス、スコア、フィードバックには影響しません。

ペイロードのプロセス内保持

capture_payloads: false、または MCPD_CAPTURE_PAYLOADS=false にすると、プロンプトと補完は その場に留まります。スパンは所要時間、ステータス、使用量、モデル、相関、 エラーを持って出ていきますが、input:、output:、 ブロックが返した値は、キューに入る前に捨てられます。機密がキューに入ることも、 フラッシュで外に出ることもありません。個々のスパンは capture_payloads: true で別の扱いにできます。

記録した内容の検証

シンクは、吐き出されたバッチの行き先です。既定は HTTP トランスポート です。Mcpd.Sink.Recorder はすべてのバッチをメモリに保持するので、 テストはデーモンなしで、計装が記録したスパン、ログ、メトリクス、スコア、 フィードバックを読み返せます。Mcpd.Sink.Console は、シグナル 1 件につき JSON を 1 行、標準エラー出力に書きます。自分のシンクは Mcpd.Sink を実装します。例外を投げるシンクはバッチの失敗として 数えられ、報告されるだけで、呼び出し側に例外が届くことはありません。

Mcpd.record/3 は、すでに終了したスパンを渡されたままキューに 入れます。id、親、開始時刻、終了時刻はそのままで、親を付け替えたり、時刻を 打ち直したり、サンプリングをやり直したりはしません。別のトレーシングシステムの アダプターや一括インポートが持っているのが、これです。id が OpenTelemetry の 形式でないレコードは受け付けられずに数えられ、例外にはなりません。どのスパン でも external_parent_span_id: は、mcpd 以外のトレーシング システムの親を、相関のためだけに運びます。

永続実行

同じクライアントがデーモンの永続実行も扱います。Mcpd.Workflow が 定義を組み立て、ステップの本体を 3 引数の関数として保持します。 Mcpd.StepWorker がそれを登録し、リースをロングポーリングで受け 取って、1 件ずつ専用のプロセスで処理します。Mcpd.start_run/3、 Mcpd.watch_run/3、Mcpd.publish_event/4 なども、 実行の API を 1 対 1 で写しています。本体は class: :retryable を 付けた Mcpd.StepException を送出するとステップの予算の範囲で リトライされ、Mcpd.StepContext.span/4 を通して自分のステップの下に スパンを記録します。本体が閉じたスパンはすべて、その使用量をステップの使用量に 足し込み、最後に見えたモデルが採用されるので、デーモンはトレースに出ている とおりに料金を計算します。set_usage/2 は、それを明示的に上書き します。ワーカーには on_error の隣に on_lease、 on_step_start、on_step_end、 on_conflict があり、StepWorker.stats/1 は取得した数、 完了した数、クラス別に失敗した数、理由別に手放した数を数えます。

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

それぞれの構成要素の働き、リースのプロトコル、実行の API は ワークフロー、 ワーカー、 実行のページにあります。

スコアとフィードバック

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

読み出し

{: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

短命なプロセスにはフラッシュが必要

すぐに終了するスクリプトやタスクは、送りきれなかったものを失います。 Mcpd.shutdown/1 はフラッシュしてワーカーを止めます。2 回呼んでも 問題ありません。

次に

• この SDK がほかの 5 つと共有している振る舞いのすべて： SDK コントラクト。
• スパンが持つもの： データモデル。
• あとから目的の実行を見つける方法： 実行を探す。