Hızlı başlangıç
Daemon çalışıyor, ajan ölçümlendi, trace ekranda. Kaydolunacak bir yer ve yanına kurulacak bir şey yok.
1. Daemon'u çalıştırın
Henüz bir sürüm yayımlanmadı.
Bu sitedeki komutların adını verdiği imaj, chart ve paketler planlanan dağıtımdır; hiçbiri yayımlanmadı. Bugün çalışan yol, kaynaktan derlemektir ve Rust ile Node gerekir:
Derleyip çalıştırma
git clone https://github.com/mcpd/mcpd && cd mcpd && make build && ./target/release/mcpd demo mcpd, depolaması içine derlenmiş tek bir süreçtir. Ayağa kaldırılacak bir veritabanı ve yazılacak bir yapılandırma dosyası yoktur.
docker run -p 4317:4317 -v mcpd-data:/data ghcr.io/mcpd/mcpd Volume isteğe bağlıdır ama eklemeye değer: onsuz, konteyner gittiğinde trace'ler de gider. İkili dosyayı mı tercih edersiniz? Doğrudan kurun.
$ docker run -p 4317:4317 ghcr.io/mcpd/mcpdWARN listening on 0.0.0.0:4317 with open ingest. Anyone who can reach this port can write to it. Set --ingest-key (or MCPD_INGEST_KEYS) before exposing it.WARN listening on 0.0.0.0:4317 with an open read API. Stored prompts and completions are readable by anyone who can reach this port. Set --read-key to require a key.WARN listening on 0.0.0.0:4317 with an open run surface. Anyone who can reach this port can start runs and claim steps. Set --execute-key and --worker-key before exposing it.INFO mcpd 0.1.0 listening on http://0.0.0.0:4317INFO interface http://0.0.0.0:4317/INFO OTLP http://0.0.0.0:4317/v1/tracesINFO native http://0.0.0.0:4317/v1/ingestINFO runs http://0.0.0.0:4317/v1/api/runsINFO storage sqlite at /data/mcpd.db Uyarılar en başta gelir ve bu bilinçlidir. Dizüstünüzde bir önemleri yok. Bu porta başka bir şey erişebilir hâle gelmeden önce Dağıtım sayfasını okuyun.
2. Ona bir şey gönderin
İki giriş yolu var ve bizimkini seçmek zorunda değilsiniz. Ajanınız zaten OpenTelemetry üretiyorsa onu mcpd'ye yöneltin ve tek satır yeni kod yazmayın:
export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 Aksi hâlde dilinizin SDK'sını kurun. Zorunlu bağımlılığı yoktur ve bir span kaydetmek asla bloklamaz, asla hata fırlatmaz.
bağlanma
import { Mcpd } from '@mcpd/sdk'
// Every field here also reads an environment variable, so the same code runs
// unchanged in development and in production. With no arguments at all this
// talks to a daemon on http://localhost:4317.
const mcpd = new Mcpd({
project: 'support',
environment: 'development',
serviceName: 'support-agent',
serviceVersion: '1.4.0',
}) 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",
) // Nothing here is required. `Mcpd::new()` reads MCPD_ENDPOINT, MCPD_API_KEY
// and the rest from the environment and talks to a local mcpd by default.
// Anything set in code wins over the environment.
let mcpd = Mcpd::with_config(
Config::from_env()
.with_project("support")
.with_environment("local")
.with_service("support-agent", Some("1.0.0")),
); m := mcpd.New(
mcpd.WithEndpoint("http://localhost:4317"),
mcpd.WithProject("support"),
mcpd.WithEnvironment("development"),
mcpd.WithService("support-agent", "1.4.2"),
)
// Shutdown flushes what is still queued and stops the background worker.
// A short-lived process that skips it loses whatever had not been sent.
defer m.Shutdown(ctx) // 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",
)
) # 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"
) # mcpd needs no registration and no project to be created up front. Check that
# the daemon is up and which spec version it implements.
curl -s "$MCPD/healthz"
# → {"status":"ok","version":"0.1.0","spec_version":"1.0.0", ...} 3. Bir çalıştırmayı ölçümleyin
Ajanınızın yaptığı işi sarın. Bir span; neyin girdiğini, neyin çıktığını, ne kadar sürdüğünü ve ne kadara mal olduğunu kaydeder. Alt span'ler onun içine kendiliğinden yerleşir, böylece bloğun içinde yapılan bir araç çağrısı, nereye ait olduğu söylenmeden ağaçta doğru yere düşer.
bir çalıştırmayı kaydetme
/**
* One ticket, one trace.
*
* `withSpan` opens the span, keeps it active for everything the callback
* starts, and ends it when the callback settles. Nothing inside has to be
* passed a handle to be nested under it.
*/
async function handleTicket(ticket: Ticket): Promise<{ answer: string; traceId: string }> {
let traceId = ''
const answer = await mcpd.withSpan('support_agent', async (run) => {
// Keep the id: a score or a piece of feedback that arrives later attaches
// to this run by it.
traceId = run.traceId
// Tags label the whole run, so the trace list can filter on them.
run.setTags(['support', 'tier-1'])
run.setCorrelation({ sessionId: ticket.id, userId: ticket.customerId })
run.setInput({ question: ticket.question })
const order = await lookUpOrder(ticket)
const answer = await draftAnswer(ticket, order)
run.setOutput({ answer })
return answer
})
return { answer, traceId }
} # 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 // The root span is the run. The sampling decision is made here, once, and
// every child inherits it.
let run = mcpd.trace("support agent");
run.set_input(question);
run.set_session("session-4f21");
run.set_user("customer-91");
run.set_tags(["support", "orders"]);
// A tool call. `run.child_of` nests it under the run.
let lookup = run.child_of("lookup_order", SpanKind::ToolCall);
lookup.set_input(json!({ "order_id": "5512" }));
let order = json!({ "status": "shipped", "carrier": "DHL", "eta": "Tuesday" });
lookup.set_output(&order);
lookup.end();
// A tool that fails. Recording the error sets the span's status to `error`,
// and a message is required: "something failed" is not actionable.
let notify = run.child_of("notify_customer", SpanKind::ToolCall);
notify.set_input(json!({ "channel": "sms" }));
notify.set_error("sms gateway timed out after 5s");
notify.end();
// The failure did not stop the run: the agent falls back to email, and the
// trace shows both the attempt and the recovery.
run.event_with("fell back to email", json!({ "after": "sms gateway timeout" })); // handleTicket is the agent. It drafts an answer with a model, tries to send
// it with a tool, and fails on the send so the error path is recorded too.
func handleTicket(ctx context.Context, m *mcpd.Client, question string) string {
ctx, run := m.Start(ctx, "support_agent",
mcpd.WithSpanKind(mcpd.SpanKindAgentRun),
mcpd.WithEntity(mcpd.EntityKindAgent, "support-agent", "Support agent"),
mcpd.WithSession("sess-4821"),
mcpd.WithUser("u-317"),
mcpd.WithTags("support", "billing"),
mcpd.WithInput(question),
)
defer run.End()
m.Log(ctx, mcpd.LogLevelInfo, "ticket received", map[string]any{"channel": "email"})
answer := draftAnswer(ctx, m, question)
run.SetOutput(answer)
if err := sendReply(ctx, m, answer); err != nil {
// The span is now an error with a message, which is what makes it
// worth reading. End below records the failed status.
run.RecordError(err, mcpd.ErrorDomain("email"), mcpd.ErrorRetryable(true))
}
return run.TraceID()
} 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
} 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 # One batch carries any mix of the five signals. Send the whole run at once.
curl -s -X POST "$MCPD/v1/ingest" \
-H 'content-type: application/json' \
-H "authorization: Bearer ${MCPD_API_KEY:-}" \
-d @- <<JSON
{
"ingest_key": "run-$TRACE",
"spans": [
{
"trace_id": "$TRACE",
"span_id": "$ROOT",
"name": "support_agent",
"kind": "agent_run",
"status": "ok",
"start_time": "$NOW", "end_time": "$NOW",
"is_root": true, "is_event": false, "received_at": "$NOW",
"input": { "question": "Where is order 4417?" },
"output": { "answer": "It is in the Leeds depot." },
"correlation": { "project": "default", "environment": "production" }
},
{
"trace_id": "$TRACE",
"span_id": "$GEN",
"parent_span_id": "$ROOT",
"name": "chat",
"kind": "model_generation",
"status": "ok",
"start_time": "$NOW", "end_time": "$NOW",
"is_root": false, "is_event": false, "received_at": "$NOW",
"model": "gpt-4o",
"provider": "openai",
"usage": { "input_tokens": 1284, "output_tokens": 210 },
"correlation": { "project": "default" }
}
]
}
JSON
# → {"accepted":2,"rejected":0,"duplicate":0}
#
# mcpd prices the model span from its own price book on the way in, so cost is
# stored with the run rather than recomputed later from rates that may have
# changed since. 4. Okuyun
localhost:4317 adresini açın. Çalıştırmanız listede. Onu açtığınızda ağacın tamamını alırsınız: her model çağrısı prompt'u ve tamamlamasıyla, her araç çağrısı argümanları ve sonucuyla, süreler, token sayıları ve maliyet.
- agent support_agent fail
- model chat
- tool returns_policy
- model chat
- tool returns_policy
- model chat
- tool returns_policy
The model called one tool three times and never answered. That is visible in four seconds here, and invisible in a log file.
5. İlginç olanı bulun
Vaktinizi listede değil, aramada geçireceksiniz. mcpd her prompt'u ve her tamamlamayı dizinler; böylece bir müşterinin şikâyet ettiği şeyi modelin söylediği çalıştırmaya doğrudan gidebilirsiniz:
arama ve analiz
// The same client reads. This is the query that finds bad answers: every run
// that a scorer put below the bar.
const suspect = await mcpd.listTraces({
hasScore: 'answer_relevance',
scoreBelow: 0.5,
limit: 10,
})
console.log(`runs scoring under 0.5: ${suspect.traces.length}`)
const recent = await mcpd.listTraces({ limit: 5, sort: 'start_time', order: 'desc' })
for (const trace of recent.traces) {
const seconds = ((trace.durationNs ?? 0) / 1e9).toFixed(2)
console.log(
`${trace.traceId.slice(0, 8)} ${trace.name.padEnd(16)} ${trace.status.padEnd(6)} ` +
`${seconds}s ${trace.spanCount} spans $${(trace.cost?.totalCost ?? 0).toFixed(6)}`,
)
}
// One run, with every span in it, ordered so a waterfall renders directly.
const detail = await mcpd.getTrace(traceId, { include: ['scores', 'feedback'] })
for (const span of detail.spans) {
console.log(` ${span.kind.padEnd(18)} ${span.name}`)
} # 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}") // The same client reads. This is the trace you just wrote.
match mcpd.get_trace(&trace_id, Include::all()).await {
Ok(detail) => {
println!("{}: {} spans", detail.trace.name, detail.spans.len());
for span in &detail.spans {
println!(
" {:<22} {:>8} {}",
span.name,
span.duration_ms().map_or("open".to_owned(), |ms| format!("{ms:.1}ms")),
span.status,
);
}
if let Some(cost) = detail.trace.cost.as_ref().and_then(|c| c.total_cost) {
println!(" cost: {cost:.6} USD");
}
for score in detail.scores.iter().flatten() {
println!(" score {}: {}", score.scorer_id, score.value);
}
}
Err(e) => eprintln!("could not read the trace back: {e}"),
} // Recording is asynchronous, so wait for the batch to land before reading
// it back. Outside an example you would never need this.
if err := m.Flush(ctx); err != nil {
log.Printf("flush: %v", err)
}
detail, err := m.GetTrace(ctx, traceID, "scores", "feedback")
if err != nil {
log.Fatalf("read the run back: %v", err)
}
fmt.Printf("%s: %s in %s, %d spans\n",
detail.Trace.TraceID, detail.Trace.Status, duration(detail.Trace.DurationNs), len(detail.Spans))
for _, span := range detail.Spans {
fmt.Printf(" %-18s %-18s %s\n", span.Kind, span.Name, span.Status)
}
summary, err := m.AnalyticsSummary(ctx, time.Now().Add(-time.Hour), time.Now(), "model")
if err != nil {
log.Fatalf("read the summary: %v", err)
}
fmt.Printf("last hour: %d runs, %.1f%% failed, %.4f %s\n",
summary.TraceCount, summary.ErrorRate*100, summary.TotalCost, summary.Currency) 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",
) {: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 # Read it back. Any filter in the API works here.
curl -s "$MCPD/v1/api/traces?project=default&limit=10"
# The whole run, with its logs, scores and feedback in one request.
curl -s "$MCPD/v1/api/traces/$TRACE?include=all"
# Find the run by something the model actually said.
curl -s --get "$MCPD/v1/api/traces" \
--data-urlencode 'project=default' \
--data-urlencode 'q=Leeds depot'
# Find the runs a scorer marked down. This is the query that finds bad answers.
curl -s --get "$MCPD/v1/api/traces" \
--data-urlencode 'project=default' \
--data-urlencode 'has_score=answer_relevance' \
--data-urlencode 'score_below=0.5'
# Where the money went, over the last week, by model.
curl -s --get "$MCPD/v1/api/analytics/cost" \
--data-urlencode 'project=default' \
--data-urlencode 'from=-7d' \
--data-urlencode 'group_by=model' Bundan sonra nereye
Artık döngünün tamamı elinizde. Sırada okumaya değer sayfalar ne yaptığınıza bağlı:
- Bunu başkaları için çalıştırmak: Docker ve Kubernetes, ardından Maskeleme ve saklama.
- Faturayı izlemek: Maliyet muhasebesi.
- Kaliteyi ölçmek: Puanlar ve geri bildirim.
- Zaten OpenTelemetry kullanıyorsanız: OTLP yolunun tamamı.