Kotlin

dependencies {
    implementation("dev.mcpd:mcpd-kotlin:0.1.0")
}

// Every field also reads an environment variable (MCPD_ENDPOINT,
// MCPD_PROJECT, MCPD_API_KEY, ...), so `Mcpd()` on its own works too.
val mcpd = Mcpd(
    McpdConfig(
        endpoint = "http://localhost:4317",
        project = "support-desk",
        environment = "development",
        serviceName = "support-agent",
        serviceVersion = "1.4.0",
    )
)

// ブロックの値がそのまま span の値になるので、計装のためにまわりのコードを
// 組み替える必要はない。
val answer = mcpd.span("retrieve_context", SpanKind.RETRIEVAL) { span ->
    span.setInput(query)
    val docs = index.search(query)
    span.setOutput(docs.map { it.id })
    span.setAttribute("retrieval.hits", docs.size)
    docs
}

val question = "where is my order 41?"

// `span` starts the operation, ends it, and records anything it throws.
// Inside the block the span is the active one, so the helpers below find
// their parent through the coroutine context rather than being handed it.
val traceId = mcpd.span(
    "support_agent",
    SpanKind.AGENT_RUN,
    input = question,
    sessionId = "session-8f21",
    userId = "customer-4477",
    tags = listOf("support", "email"),
) { run ->
    mcpd.log(LogLevel.INFO, "handling a question", mapOf("channel" to "email"))

    val plan = planAnswer(mcpd, question)
    val order = lookUpOrder(mcpd, 41)
    val estimate = estimateDelivery(mcpd, 41)

    val answer = buildString {
        append(plan).append(' ').append(order)
        if (estimate != null) append(" It should arrive ").append(estimate).append('.')
    }
    run.setOutput(answer)
    mcpd.metric("support.answers", 1.0, MetricKind.COUNTER, unit = "{answer}")

    // The block's value is the value of `span`.
    run.traceId
}

val run = mcpd.trace("nightly_reindex", SpanKind.WORKFLOW_RUN, tags = listOf("batch"))
try {
    reindex()
    run.end()
} catch (e: Exception) {
    run.setError(e)
    run.end()
}

mcpd.span("support_agent", SpanKind.AGENT_RUN) { run ->
    // draft には何も渡していない。何段 suspend を下っても、コルーチン
    // コンテキストから実行を見つける。
    run.setOutput(draft(question))
}

suspend fun draft(question: String): String =
    mcpd.span("draft_answer", SpanKind.MODEL_GENERATION) { generation ->
        generation.setInput(question)
        "..."
    }

// 送信側: ヘッダーを次のサービスに渡す。
request.header("traceparent", span.traceparent())

// 受信側: 新しいトレースを始めず、そのトレースを続ける。
mcpd.span("checkout", traceparent = headers["traceparent"]) { span ->
    charge(span)
}

generation.setModel("gpt-4o", provider = "openai")
generation.setUsage(inputTokens = 1_180, outputTokens = 96)

// setError はすでに持っている throwable を取るので、メッセージ、クラス、
// スタックはすべてそこから取れる。
tool.setError(e, retryable = true)

// あるいはメッセージを直接。どちらにしてもメッセージは必須。
tool.setError("the shipping service returned 503")

// プロンプトと補完はプロセスの中に留める。スパンは所要時間、ステータス、
// 使用量、モデル、相関、エラーをそのまま持つ。
val quiet = Mcpd(McpdConfig(capturePayloads = false))
quiet.span("draft_answer", input = question, capturePayloads = true) { }  // これだけは残す

// 別のトレーサーがすでに終了させたスパン。渡されたままキューに入り、親の
// 付け替えも、時刻の打ち直しも、サンプリングのやり直しもしない。形式の壊れた
// id は捨てられて数えられるだけで、例外にはならない。
mcpd.record(finishedSpan)

// サーバーなしで、記録された内容を確かめる。
val recorder = RecordingSink()
val local = Mcpd(McpdConfig(sink = recorder))
local.span("support_agent") { run -> run.setUsage(inputTokens = 12, outputTokens = 3) }
local.flush()
check(recorder.spans.single().usage?.inputTokens == 12L)

// queued, dropped, capacity, exported, failed, rejected.
val stats = local.stats()

val support = Workflow("support_ticket")
    .step("classify") { ctx ->
        val ticket = ctx.input<JsonObject>()["ticket"]!!.jsonPrimitive.content
        mapOf("kind" to if (ticket.startsWith("Q")) "question" else "bug")
    }
    .agent("draft", StepOptions(input = mapOf("kind" to "\${steps.classify.output.kind}"))) {
        ctx ->
        // Your own model call, recorded under this step in the run's trace.
        // Its usage is summed into the step, so the daemon prices the run.
        ctx.span("model_call", SpanKind.MODEL_GENERATION) { span ->
            span.setModel("claude-sonnet-4", "anthropic")
            span.setUsage(inputTokens = 812, outputTokens = 37)
        }
        "an answer to a " + ctx.input<JsonObject>()["kind"]!!.jsonPrimitive.content
    }
    .waitFor(
        "approval", "approved", "\${input.ticket}", 7.days,
        StepOptions(guard = predicate("steps.classify.output.kind", "eq", "bug")),
    )
    .step(
        "send",
        StepOptions(
            input = mapOf("text" to "\${steps.draft.output}"),
            effect = Effect.AT_MOST_ONCE,
            retry = RetryPolicy(
                maxAttempts = 3, backoff = Backoff.EXPONENTIAL, initial = 500.milliseconds,
            ),
        ),
    ) { ctx ->
        val text = ctx.input<JsonObject>()["text"]!!.jsonPrimitive.content
        if (!deliver(text, ctx.idempotencyKey)) {
            throw RetryableException("mail relay busy", retryAfter = 2.seconds)
        }
        mapOf("sent" to true)
    }
    .output(mapOf("sent" to "\${steps.send.output.sent}"))

// 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(
    traceId = traceId,
    scorerId = "answer_relevance",
    value = 0.86,
    min = 0.0,
    max = 1.0,
    reason = "answered the question asked, without the delivery estimate",
)

// A person's judgement, from the thumbs-up control in your own product.
mcpd.feedback(
    traceId = traceId,
    kind = FeedbackKind.THUMBS,
    value = "up",
    comment = "fast and clear",
    authorId = "customer-4477",
)

// Nothing above touched the network. This is where it goes.
mcpd.flush()

// JVM のサービス: 終了時にシャットダウンする。
Runtime.getRuntime().addShutdownHook(Thread { mcpd.close() })

// 短命なワーカー: use を使えば、閉じ括弧でシャットダウンされる。
Mcpd(McpdConfig(project = "batch")).use { mcpd ->
    mcpd.span("reindex") { run -> reindex(run) }
}

val detail = mcpd.getTrace(traceId, include = listOf("scores", "feedback"))
println("${detail.trace.name}: ${detail.spans.size} spans, ${detail.trace.status}")
for (span in detail.spans) {
    val ms = (span.durationNs ?: 0) / 1_000_000
    println("  ${span.kind} ${span.name} ${ms}ms ${span.error?.message ?: ""}")
}

val summary = mcpd.analyticsSummary(
    from = Clock.System.now() - 1.hours,
    to = Clock.System.now(),
)
println(
    "last hour: ${summary.traceCount} runs, " +
        "${summary.errorCount} failed, ${summary.totalCost} USD",
)