Rust

Kurulum

SDK dışa aktarımı arka plandaki bir tokio görevinde yapar, yani bir runtime bulmayı bekler. Runtime olmadan da çalışır; o zaman siz flush() çağırana kadar hiçbir şey gönderilmez. Bu, senkron bir araç için doğru davranıştır ama uzun ömürlü bir servisin isteyeceği şey değildir.

İstemciyi oluşturma

Süreç başına bir istemci. Mcpd::new() ortamı okur ve hiçbir şey ayarlanmamışken yerel bir mcpd ile konuşur. Oluşturma hiçbir zaman başarısız olmaz: mcpd'nin kullanamayacağı bir yapılandırma istemciyi devre dışı bırakır ve nedenini loglar; çünkü bir gözlemlenebilirlik çağrısı, bir programın hatasını ele almak zorunda kalacağı bir şey değildir.

// 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")),
);

Config::from_env() çağrısından sonra builder üzerinde verilen her şey ortama üstün gelir.

Ayrıştırılamayan bir ayar bir uyarıyla atlanır ve varsayılan kullanılır; çünkü bir dağıtımın ortamındaki bir yazım hatası programı durdurmamalıdır. enabled = false her çağrıyı, yine de geçerli nesneler döndüren boş bir işleme çevirir; kimlikler yine düzgün biçimlidir ve traceparent yine taşınır, yani bir test sınadığı koda dokunmadan mcpd'yi kapatabilir.

Bir çalıştırmayı kaydetme

Span bir RAII koruyucusudur. Drop edildiğinde sona erer; Rust'ta kapsamlı biçim budur: bir blok, bir kapsam ya da fonksiyonun sonu. end() çağrısını kendiniz yapmak da aynı şeyi yapar. Koruyucunun varlık nedeni, erken bir dönüşün ya da bir paniğin bir span'i açık ve bitmemiş bırakamamasıdır.

let run = mcpd.trace("support agent");
{
    let step = run.child_of("retrieve", SpanKind::Retrieval);
    step.set_input("order 5512");
    // Bu adım ne yapıyorsa. Span kapanış süslü parantezinde biter; erken bir
    // dönüşte, bir ? işaretinde ve bir panikte de biter.
}
run.end();

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

Bağlam

Bir sürecin içinde bağlam, bir tokio task-local'ında taşınır. Bir span'i elde tutmak onu geçerli span yapmaz: Span::scope task-local'ı bir future için, Span::sync_scope ise bir closure için ayarlar. İkisinin de içinde Mcpd::span sürmekte olan çalıştırmaya bağlanır, yani birkaç katman aşağıdaki bir yardımcı fonksiyon, üst span kendisine verilmeden doğru yere yerleşir. Üst span zaten elinizdeyse Span::child_of açık seçenektir.

// Bir span'i açmak onu geçerli span yapmaz. scope() yapar.
let run = mcpd.trace("support agent");
run.scope(async {
    lookup(&mcpd).await;
})
.await;
run.end();

async fn lookup(mcpd: &Mcpd) {
    // Bu fonksiyona hiçbir üst geçirilmedi, yine de çalıştırmanın altına
    // yerleşir; çünkü onu task-local taşır.
    let step = mcpd.span("lookup_order", SpanKind::ToolCall);
    step.set_output(serde_json::json!({ "status": "shipped" }));
}

Süreçler arasında bağlam, W3C traceparent başlığında taşınır. Örnekleme bayrağı da onunla birlikte gider, böylece bir trace'in iki yarısı trace'in saklanıp saklanmadığı konusunda uyuşur.

// Giden: başlığı isteğin üzerine koyun.
let header = span.traceparent();

// Gelen: yeni bir trace başlatmak yerine çağıranın trace'ini sürdürün.
// Bulunmayan ya da bozuk bir başlık, başarısız olmak yerine yeni bir trace başlatır.
let run = mcpd.trace_from("charge card", inbound_header, SpanKind::AgentRun);

Model, kullanım ve hatalar

mcpd alım sırasında token sayılarını fiyatlandırır, dolayısıyla trace'te gördüğünüz maliyet size faturalanan maliyettir. Bir hatayı kaydetmek aynı hamlede span'in durumunu error yapar, yani ikisi hiçbir zaman çelişemez; mesaj da seçime bağlı değil, zorunlu bir argümandır.

generation.set_model("claude-sonnet-4", "anthropic");
generation.set_usage(TokenUsage {
    input_tokens: Some(412),
    output_tokens: Some(96),
    ..TokenUsage::default()
});
// Ya da döküm önemli değilse:
generation.set_tokens(412, 96);

// Mesaj zorunludur. "Bir şey başarısız oldu" ile bir şey yapılamaz.
tool.set_error("sms gateway timed out after 5s");

// Ya da elinizdeki hatayı kaydedin; mesajını ve türünü ondan okur.
tool.record_error(&err);

Geri kalan ayarlayıcılar şunlardır: set_input, set_output, set_attributes, set_metadata, set_tags, set_entity, set_session, set_user, set_status ve set_error_detail. Hepsi &self alır, yani bir span ödünç alındığı her yerden doldurulabilir.

Payload'lar, sink'ler ve bitmiş span'ler

Config::with_capture_payloads(false) prompt'ları ve tamamlamaları sürecin içinde tutar: span'ler yine zamanlamasını, durumunu, token kullanımını, modelini, ilişkilendirmesini ve hatalarını taşır, input ile output ise kuyruğa varmadan atılır. Bir span, span.capture_payloads(..) ile istemcinin kararını kendisi için iki yönde de geçersiz kılar.

Exporter her yığını istemcinin sink'ine verir. Varsayılan olan HTTP taşıyıcısıdır; Config::with_sink, yığınları bellekte tutan ve böylece bir testin kaydedileni sunucu olmadan doğrulamasını sağlayan bir RecordingSink ya da sinyal başına tek bir JSON satırı yazan bir ConsoleSink alır. Başarısız olan bir sink, aralık başına bir kez bildirilen ve asla fırlatılmayan başarısız bir dışa aktarımdır. mcpd.stats() şunları yanıtlar: queued, dropped, capacity, exported, failed ve rejected.

mcpd.record(span), çoktan bitmiş bir span'i kimlikleri, üstü, başlangıcı ve bitişiyle birlikte, bir adaptörün ya da bir içe aktarmanın elinde nasılsa öyle kuyruğa alır: üstü, zamanları ve örneklemesi hiç değiştirilmeden. Bozuk bir kayıt atılır ve rejected olarak sayılır. span.set_external_parent, mcpd olmayan bir izleme sisteminden gelen bir üstü yalnızca ilişkilendirme için taşır.

Kalıcı çalıştırmalar

Daemon'ın kalıcı çalıştırmalarını da aynı istemci karşılar. Workflow::new tanımı kurar ve adım gövdelerini async closure olarak saklar; bir Worker bu tanımı kaydeder, lease'ler için uzun yoklama yapar ve her birini bir görev üzerinde karşılar; start_run, watch_run, publish_event ve diğerleri çalıştırma yüzeyini bire bir yansıtır. Bir gövde, adımın bütçesi içinde yeniden denenmek için StepFailure::retryable döndürür ve StepContext üzerinden kendi adımının altına span kaydeder. Böyle açılan her span ve altına yerleşen her span, bittiğinde kendi kullanımını adıma ekler; böylece daemon, gövde token'ları elle toplamadan trace'in gösterdiğini fiyatlandırır. set_usage açık geçersiz kılmadır. Builder'ın on_lease, on_step_start, on_step_end ve on_conflict hook'ları ile worker.stats(), worker'ı içeriden gözlemlenebilir kılar.

let support = Workflow::new("support_ticket")
    .step(
        "classify",
        |ctx: StepContext| async move {
            let input: Value = ctx.input()?;
            let ticket = input["ticket"].as_str().unwrap_or("");
            let kind = if ticket.starts_with('Q') { "question" } else { "bug" };
            Ok::<_, StepFailure>(json!({ "kind": kind }))
        },
        StepOptions::new(),
    )
    .agent(
        "draft",
        |ctx: StepContext| async move {
            // Your own model call, recorded under this step in the run's trace.
            // Its usage is the step's usage: the daemon prices what the span spent.
            let span = ctx.span("model_call", SpanKind::ModelGeneration);
            span.set_model("claude-sonnet-4", "anthropic");
            span.set_tokens(812, 37);
            span.end();
            Ok::<_, StepFailure>("an answer")
        },
        StepOptions::new().input(draft_input),
    )
    .wait_for(
        "approval",
        "approved",
        "${input.ticket}",
        Duration::from_secs(7 * 24 * 3600),
        StepOptions::new().when(when("steps.classify.output.kind", "eq", "bug")),
    )
    .step(
        "send",
        |ctx: StepContext| async move {
            let input: Value = ctx.input()?;
            let text = input["text"].as_str().unwrap_or("");
            deliver(text, ctx.idempotency_key()).await.map_err(busy)?;
            Ok::<_, StepFailure>(json!({ "sent": true }))
        },
        StepOptions::new()
            .input(json!({ "text": "${steps.draft.output}" }))
            .at_most_once()
            .retry(backoff),
    )
    .output(json!({ "sent": "${steps.send.output.sent}" }));

Her yapının ne yaptığı, lease protokolü ve çalıştırma API'si şu sayfalarda: İş akışları, Worker'lar ve Çalıştırmalar.

Puanlar ve geri bildirim

Scale zorunlu bir argümandır, seçime bağlı değil. Aralığı bildirilmemiş çıplak bir 0.9 hiçbir şeyle karşılaştırılamaz ve biri grafiği okuduğunda bunun ne anlama geldiğini sormak için artık çok geçtir. Alışılmış olanı, 0.0 ile 1.0 arasındaki Scale::UNIT. Geri bildirim bir insanın yargısıdır ve puandan ayrı tutulur; çünkü ikisi farklı sorulara yanıt verir ve asla birlikte ortalanmamalıdır.

// A score is a machine judgement. The scale is not optional: a bare 0.9 with
// no declared range cannot be compared to anything.
mcpd.score(
    Target::trace(&trace_id),
    "answer_relevance",
    0.9,
    Scale::UNIT,
    Some("Names the carrier and the delivery day the customer asked about."),
);

// Feedback is a person's judgement, kept separate from a score because the
// two answer different questions and must never be averaged together.
mcpd.feedback(Target::trace(&trace_id), FeedbackKind::Thumbs, "up", Some("fast and correct"));

Flush ve kapatma

Kayıt, bellekteki bir kuyruğa ekler ve döner. flush() kuyrukta bekleyeni gönderir; shutdown() flush eder, worker'ı durdurur ve iki kez çağrılması güvenlidir. İkisi de beklenebilir ve ikisi de son dışa aktarım hatasını döndürür, böylece onu loglayabilirsiniz.

Kısa ömürlü bir süreç bunlardan birini çağırmalıdır. Kapatmadan çıkan bir ikili dosya, span'leri hâlâ bellekteyken çıkar; serverless bir fonksiyonda ise runtime, handler döner dönmez süreci dondurur, yani arka plandaki göreve bir daha sıra gelmez.

#[tokio::main]
async fn main() {
    let mcpd = Mcpd::new();

    // ... çalıştırma ...

    // Bunu yok saymak makul bir varsayılandır: bir gözlemlenebilirlik hatası
    // uygulama hatası değildir. Loglamak bir satıra mal olur.
    if let Err(e) = mcpd.shutdown().await {
        eprintln!("mcpd could not export: {e}");
    }
}

Başarısız bir flush yüzünden kaybolan sinyaller yeniden kuyruğa alınmaz. Taşıyıcı zaten yeniden denemiştir; başarısızlıkta kendini yeniden dolduran bir kuyruk da bir kesintinin bellek yetersizliğinden ölüme dönüşme yoludur.

Güvenebilecekleriniz

Daemon durdurulmuşken örnek program sonuna kadar çalışır, yapamadığı şeyi yazar ve sıfır ile çıkar.

$ cargo run -p mcpd-sdk --example quickstart    Finished `dev` profile [unoptimized + debuginfo]     Running `target/debug/examples/quickstart`mcpd could not export: export failed: mcpd could not be reachedcould not read the trace back: mcpd could not be reached$ echo $?0

• Hiçbir ölçümleme çağrısı hata döndürmez ve panik etmez. Erişilemeyen bir mcpd, bir serileştirme hatası ya da dolu bir kuyruk, flush aralığı başına en çok bir kez tracing üzerinden loglanır ve yutulur. Gözlemlenebilirlik, gözlediği şeyi asla devirmemelidir.
• Kayıt hiçbir zaman bloklamaz. Dışa aktarım arka plandaki bir tokio görevinde olur, yani ajanınızın gecikmesi mcpd'nin hızlı ya da ayakta olmasına bağlı değildir.
• Kuyruk max_queue ile sınırlıdır. Dolduğunda en eski sinyaller atılır ve mcpd.stats().dropped bunları sayar. En eskiyi atmak, bellek yetersizliğinden ölmekten daha küçük bir başarısızlıktır.
• 1.0 altındaki örnekleme rastgele değil, trace id'sine göre belirlenimlidir; böylece bir filodaki her süreç aynı trace hakkında aynı kararı verir ve servis sınırını aşan bir trace hiçbir zaman yarım kaydedilmez.
• Bir span başladığında dışa aktarılır, bittiğinde güncellenir; böylece yolun ortasında çöken bir çalıştırma yine de okunabilir kalır.

Geri okuma

Okumayı da aynı istemci yapar. Her metot bir HTTP endpoint'ini yansıtır ve üretilmiş tipleri döndürür; bu tipler spesifikasyondan mcpd-core yoluyla gelir ve mcpd_sdk::model olarak yeniden dışa aktarılır. Okumalar bir Result döndürür, çünkü beklediğiniz ve yanıtlanamayan bir sorgu haber değeri taşır.

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

Çalıştırma bulma filtreleri eksiksiz anlatır. Veri modeli bu tiplerin taşıdığı her alanı listeler.