Python

Kurulum

Python 3.9 veya üstü. Zorunlu bağımlılık yok, bilerek: exporter standart kütüphanedeki urllib ve gzip ile çalışır, yani mcpd'yi kurmak bir lockfile'a hiçbir şey eklemez ve ajanınızın zaten bağlı olduğu şeylerle çakışamaz.

httpx extra'sı yalnızca bağlantıların yeniden kullanımını ekler, başka bir şey değil. SDK onsuz da tıpatıp aynı davranır.

İstemciyi oluşturma

Süreç başına bir istemci. Her argüman bir ortam değişkenini de okur, böylece aynı kod tek bir düzenleme olmadan hem geliştirmede hem üretimde çalışır. Argümansız çağrılan Mcpd(), http://localhost:4317 adresindeki bir daemon ile konuşur.

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

Kod ortam değişkenine, ortam değişkeni de varsayılana üstün gelir.

enabled=False, her kayıt çağrısını hiçbir şey yapmayan ama yine de geçerli nesneler döndüren bir çağrıya çevirir; böylece sınadığınız kodu değiştirmeden testlerde mcpd'yi kapatabilirsiniz. capture_payloads=False prompt'ları ve tamamlamaları sürecin içinde tutar: span'ler yine zamanlaması, durumu, token kullanımı, modeli, ilişkilendirmesi ve hatalarıyla kaydedilir, input ile output ise kuyruğa varmadan atılır. Bir span, başlangıç alanlarındaki capture_payloads=True ile kendisi için tersini söyleyebilir; ters yönde de aynısı geçerlidir.

Bir çalıştırmayı kaydetme

mcpd.trace(name) bir çalıştırma başlatır, mcpd.span(name) ise o an etkin olan neyse onun altında bir adım başlatır. İkisi de bağlam yöneticisidir: blok çıkarken span'i kapatır, blok istisna fırlattıysa onu kaydeder ve istisnayı olduğu gibi geçirir. Gözleniyor olmak kendi hata işleyişinizi değiştirmez.

# 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

with bloğu kullanmıyorsanız run.end() çağrısını kendiniz yapın. Span her iki durumda da başladığında dışa aktarılır, yani hiç bitmeyen bir span, hiç bitmemiş bir span olarak görünür; aradığınız genellikle odur.

# Senkron ve asenkron aynı nesnedir. İkisi de çıkarken span'i kapatır ve
# blok istisna fırlattıysa onu kaydeder.
with mcpd.trace("nightly_summariser") as run:
    with mcpd.span("fetch", kind=SpanKind.TOOL_CALL) as step:
        step.set_output({"documents": 42})

async with mcpd.trace("agent") as run:
    async with mcpd.span("generate", kind=SpanKind.MODEL_GENERATION) as generation:
        generation.set_output(await model.complete(prompt))
    await mcpd.aflush()

Bağlam

Etkin span bir contextvars.ContextVar değeridir. Bir await ile göreve kadar izler ve thread'ler arasında sızmaz; böylece üç çerçeve aşağıdaki bir yardımcı fonksiyon, çağıranların hiçbiri ona argüman geçirmeden doğru üst span'in altında bir alt span açar.

def draft(question: str) -> str:
    # İçeriye hiçbir şey geçilmedi. Span yine de çalıştırmanın altına yerleşir,
    # çünkü etkin span bir contextvars.ContextVar ve bu çağrı bloğun içinde.
    with mcpd.span("draft", kind=SpanKind.MODEL_GENERATION) as generation:
        generation.set_input(question)
        return "..."


with mcpd.trace("support_agent") as run:
    run.set_output(draft("where is my order?"))

Ortaya çıkan şey, zamanlamaları ve payload'ları iliştirilmiş ağaçtır.

Servisler arasında bağlam, W3C traceparent başlığında taşınır. Bozuk bir başlık, onu taşıyan isteği başarısız kılmak yerine yeni bir trace başlatır. Gelen isteğin örneklememe kararına uyulur; çünkü bazı servislerin sakladığı, bazılarının attığı bir şey trace değildir.

# Giden: etkin span'i bir sonraki servise verin.
requests.post(url, headers={"traceparent": mcpd.traceparent()}, json=payload)

# Gelen: yeni bir trace başlatmak yerine o trace'i sürdürün.
with mcpd.trace("handler", traceparent=request.headers.get("traceparent")) as run:
    ...

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. Sağlayıcının size bildirdiğini raporlayın. set_error bir dizge ya da bir istisna alır, aynı hamlede status alanını error yapar ve her zaman bir mesaj doldurur: "bir şey başarısız oldu" ile kimse bir şey yapamaz.

generation.set_model("gpt-4o-mini", provider="openai")
generation.set_usage(input_tokens=812, output_tokens=96, cache_read=128)

# Kendiniz yakalayıp kararını kendiniz verdiğiniz bir hata.
generation.set_error("the provider timed out", kind="TimeoutError", retryable=True)

# Span içinde bir an; kendine ait bir süresi yoktur.
generation.event("first_token")

Diğer ayarlayıcılar şunlardır: set_input, set_output, set_metadata, set_attributes, set_tags, set_entity, set_cost, set_status ve add_link. Hepsi span'i döndürür ve hepsi span başlarken anahtar sözcük olarak da verilebilir. external_parent_span_id=, mcpd olmayan bir izleme sisteminden gelen bir üstü adlandırır; yalnızca ilişkilendirme için taşınır ve üstlük kurmak için hiç kullanılmaz.

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

Sink, boşaltılan bir yığının gittiği yerdir. Varsayılan olan HTTP taşıyıcısıdır. RecordingSink her yığını bellekte tutar, böylece bir test ölçümlemenin ne kaydettiğini sunucu olmadan doğrular; ConsoleSink ise sinyal başına tek bir JSON satırını stderr'e yazar. Hata fırlatan bir sink, sayılan ve bildirilen başarısız bir yığın demektir, kodunuzda asla bir istisna değil. stats(), istemcinin oluşturulmasından bu yana ne yaptığını söyler.

from mcpd import Mcpd, RecordingSink

sink = RecordingSink()
mcpd = Mcpd(sink=sink)
with mcpd.trace("answer") as run:
    run.set_usage(input_tokens=12)
mcpd.flush()
assert sink.spans[0].usage.input_tokens == 12
print(mcpd.stats())  # queued, dropped, capacity, exported, failed, rejected

mcpd.record(span), çoktan bitmiş bir SpanRecord kaydını kuyruğa alır; kimlikler, üst ve zamanlar sizde nasılsa öyle: başka bir izleme sistemi için yazılmış bir adaptörün ya da toplu bir içe aktarmanın elindeki gibi. Üstü, zamanları ve örneklemesi hiç değiştirilmez; bozuk bir kayıt atılır ve stats().rejected içinde sayılır, hiç fırlatılmaz.

Kalıcı çalıştırmalar

Daemon'ın kalıcı çalıştırmalarını da aynı istemci karşılar. workflow() tanımı kurar ve adım gövdelerini çağrı ya da dekoratör olarak saklar; bir Worker bu tanımı kaydeder, bir thread havuzunda lease'ler için uzun yoklama yapar ve her birini karşılar; mcpd.runs, mcpd.events ve mcpd.definitions ise ç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 class_="retryable" ile StepFailure fırlatır ve StepContext üzerinden kendi adımının altına span kaydeder. Bu span'lerin harcadığı, adımın kullanımıdır: her span bittikçe token'lar toplanır ve en son görülen model geçerli olur, böylece daemon adımı trace'in gösterdiğine göre fiyatlandırır; ctx.set_usage bunu alan alan geçersiz kılar. Worker, on_error yanında dört hook alır (on_lease, on_step_start, on_step_end, on_conflict) ve worker.stats() alınan lease'leri, yolda olan gövdeleri, tamamlananları, sınıfına göre başarısız olanları, nedenine göre çakışmaları ve karşılanamayan lease'leri sayar.

support = workflow("support_ticket")


@support.step("classify")
def classify(ctx, input, prior):
    return {"kind": "question" if input["ticket"].startswith("Q") else "bug"}


@support.agent("draft", input={"kind": "${steps.classify.output.kind}"})
def draft(ctx, input, prior):
    # Your own model call, recorded under this step in the run's trace. What
    # the span spends is the step's usage: the daemon prices it from here.
    with ctx.span("model_call", kind="model_generation") as span:
        span.set_model("claude-sonnet-4", "anthropic")
        span.set_usage(input_tokens=812, output_tokens=37)
        text = f"an answer to a {input['kind']}"
    return text


support.wait_for("approval", event="approved", key="${input.ticket}",
                 timeout_ms=7 * 24 * 3_600_000,
                 when=when("steps.classify.output.kind", "eq", "bug"))


@support.step("send", input={"text": "${steps.draft.output}"}, effect="at_most_once",
              retry=RetryPolicy(max_attempts=3, backoff="exponential", initial_ms=500))
def send(ctx, input, prior):
    if not deliver(input["text"], ctx.idempotency_key):
        raise StepFailure("mail relay busy", class_="retryable", retry_after_ms=2_000)
    return {"sent": True}


support.output({"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

min ve max yalnızca anahtar sözcükle verilir ve varsayılanları yoktur. Ölçeği bildirilmemiş çıplak bir sayı hiçbir şeyle karşılaştırılamaz, bu yüzden ölçek sonradan tahmin edilmek yerine çağrı yerinde zorunlu kılınır. Geri bildirim bir insanın yargısıdır ve puandan ayrı durur: ikisi farklı sorulara yanıt verir ve asla birlikte ortalanmamalıdır.

# A machine evaluation. `min` and `max` are required: a bare number with no
# declared scale cannot be compared to anything.
mcpd.score(
    "answer_relevance",
    value=round(random.uniform(0.6, 0.95), 2),
    min=0.0,
    max=1.0,
    trace_id=trace_id,
    reason="named the order and gave a delivery window, without the live lookup",
)

# A human evaluation. Kept separate from a score, because a person's
# judgement and a model's must never be averaged together.
mcpd.feedback(
    FeedbackKind.THUMBS,
    "up",
    trace_id=trace_id,
    comment="fine given the outage",
    author_kind="end_user",
)

Bir trace vermezseniz ikisi de etkin span'in trace'ine bağlanır. Loglar ve metrikler de aynısını yapar, böylece bir grafikteki sıçrama onu doğuran çalıştırmayı açar.

Flush ve kapatma

Kayıt, bellekteki bir kuyruğa ekler ve döner; HTTP işini bir daemon thread'i yapar. flush() bekleyeni gönderir ve süresi içinde gönderemezse False döndürür. shutdown() flush eder, worker'ı durdurur ve taşıyıcıyı serbest bırakır. İkisi de idempotenttir. Bir olay döngüsü içinde aflush() ve ashutdown() çağırın; bunlar aynı işi döngüyü bloklamadan yapar.

Serverless bir fonksiyonda bu, her yerden daha çok önem taşır. Runtime, handler'ınız döner dönmez süreci dondurur ve worker thread'i bir daha çalışmaz, yani hâlâ kuyrukta bekleyen bir yığın kaybolur.

def handler(event, context):
    with mcpd.trace("answer") as run:
        run.set_input(event["body"])
        answer = agent(event["body"])
        run.set_output(answer)

    # Yığın burada hâlâ bellekte. Runtime, handler döner dönmez bu süreci
    # dondurur ve worker thread'i bir daha çalışmaz.
    mcpd.flush()
    return {"statusCode": 200, "body": answer}

İstemci ayrıca bir atexit hook'u kaydeder; böylece kapatmayı unutan sıradan bir betik yine de son yığınını gönderir. Bu hook os._exit() çağrısında veya ölümcül bir sinyalde çalışmaz, yani bir emniyet ağıdır, kapatmanın yerine geçmez.

Güvenebilecekleriniz

• Ölçümleme hiçbir zaman istisna fırlatmaz. Her kayıt çağrısı korumalıdır; bir başarısızlık aralık başına en çok bir kez loglanır ve yutulur. Gözlemlenebilirlik, gözlediği şeyi asla devirmemelidir. Okuma çağrıları bunun istisnasıdır ve hata fırlatır, çünkü yanıtını beklediğiniz bir sorgu başka bir şeydir.
• Kayıt bloklamaz. Ajanınızın gecikmesi mcpd'nin hızlı ya da ayakta olmasına hiç bağlı değildir.
• Bellek sınırlıdır. max_queue kadar sinyal, varsayılan olarak 8192. Kuyruk dolduğunda en eskiler atılır, mcpd.dropped bunları sayar ve aralık başına bir uyarı loglanır. Sınırsız bir tampon, bir ağ kesintisini bellek yetersizliğinden ölüme çevirir.
• Örnekleme kararı bir kez, kökte, trace id'sinden verilir ve her alt span'e kalıtılır. Gelen bir traceparent her zaman üstün gelir.
• Bir span başladığında dışa aktarılır, bittiğinde güncellenir; böylece yolun ortasında çöken bir çalıştırma da görünür kalır.
• Zaman damgaları time.time_ns() çağrısından gelir ve hat üzerinde dokuz haneli UTC olarak gider. Yerel saat, dağıtık bir trace'i okunamaz hâle getirir.

Geri okuma

Okuma metotları HTTP API'yi yansıtır ve üretilmiş tipleri döndürür. Senkrondurlar ve başarısızlıkta McpdError fırlatırlar. from_ ile to, bir datetime ya da -15m, -1h, -7d gibi bir kısayol kabul eder. Asenkron kodda bunları asyncio.to_thread ile döngünün dışında çalıştırın.

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

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