accugents-sdk

AccuGents Python SDK

The installable Python SDK for instrumenting existing Agent applications and uploading OpenTelemetry/OpenInference-compatible execution traces to AccuGents. The PyPI distribution name is accugents-sdk; the stable Python import package remains accugent_sdk. Use is limited to authorized customers under the included proprietary LICENSE.

Release

Production publishing uses GitHub Actions and PyPI Trusted Publishing. Configure the pending publisher in PyPI with these exact values:

Restrict the pypi GitHub Environment to sdk-v* tags and disable administrator bypass. Enable required reviewer approval when the organization plan supports it. Releases use tags in the form sdk-v<version>, where <version> exactly matches project.version in pyproject.toml. Publishing a matching GitHub Release validates the SDK, builds and smoke-tests its wheel and source distribution, and publishes them without a stored PyPI token.

Implemented boundary

• immutable, unknown-field-rejecting trace identity and span records;
• agent.run, llm.call, tool.call, retrieval, citation, evaluator.judge, and error span kinds;
• OpenTelemetry-sized lowercase hexadecimal trace and span identifiers;
• ordered sequence, aware timestamps, status, attributes, and optional content;
• metadata-only content capture by default;
• category-level capture opt-in;
• recursive sensitive-key, configured-path, known-secret, and oversized-value redaction before transmission;
• bounded in-process AnyIO queue with immediate accepted/dropped submission;
• 256 KiB span guard and byte-aware batches capped at 900,000 encoded span bytes, leaving envelope headroom below the Backend's 1 MiB request limit;
• ordered batches, run-scoped flush receipts, bounded per-attempt timeout and retry, truthful close-time upload results, and delivery/loss counters;
• provider-neutral asynchronous batch transport protocol;
• strict versioned ingestion payload with retry-stable batch identity, UTC timestamp, nonce, ordered events, and deterministic canonical JSON bytes;
• secret-safe SDK write credential and domain-separated HMAC-SHA256 request signing with constant-time verification;
• optimized httpx2 HTTP/2 transport that maps signed request bodies and integrity headers to typed ingestion POST outcomes;
• public sync/async Agent-run and child-span decorators;
• context-manager instrumentation with generated run/trace/span identity, context-local parent propagation, safe error events, and run-end flush.
• pure-ASGI FastAPI middleware with exact route/method selection, streaming lifecycle evidence, downstream child-span propagation, and fail-open collection.
• minimal LangGraph wrapper for sync/async compiled-graph execution with one Agent-run root and downstream explicit child-span propagation.

The Backend exposes a production-wired signed ingestion endpoint at POST /api/v1/ingestion/batches. It resolves encrypted scoped credentials, serializes revocation, prevents nonce replay, persists retry-stable batch identity, and durably stages backend-redacted ordered events before returning 202. Terminal runs become ready for the Backend's separate transactional Evaluation projector, which preserves canonical SDK event semantics and deterministic identities. The Backend lifespan automatically drains ready projection backlog and wakes after successful ingestion commits while 202 remains a staging-only acknowledgement.

Credential onboarding and connection check

Install the SDK in the Target Agent project:

uv add accugents-sdk

Copy the one-time credential values from AccuGents into the Target Agent project root's .env file. The SDK reads this file but never creates or modifies it:

ACCUGENT_ENDPOINT=https://api.accugent.example/api/v1/ingestion/batches
ACCUGENT_KEY_ID=<issued-key-id>
ACCUGENT_SECRET_BASE64=<issued-secret>
ACCUGENT_WORKSPACE_ID=<issued-workspace-id>
ACCUGENT_PROJECT_ID=<issued-project-id>
ACCUGENT_AGENT_ID=<your-agent-id>
ACCUGENT_AGENT_VERSION=<your-agent-version>
ACCUGENT_PROMPT_VERSION=USER_INPUT_OPTIONAL

Replace every required placeholder. ACCUGENT_PROMPT_VERSION may keep USER_INPUT_OPTIONAL, be replaced with a valid prompt version, or be omitted. Then run this command from that same project root:

uv run accugent doctor

The command validates the local settings and sends one signed request to POST /api/v1/ingestion/connection-tests. It does not submit a real trace. Success prints AccuGents connection: connected; configuration, authentication, network, response, and server failures exit non-zero without printing the credential secret.

FastAPI middleware

Install the optional framework extra:

uv add "accugents-sdk[fastapi]"

Register the pure-ASGI adapter around Agent endpoints after constructing the shared Instrumentation instance:

from accugent_sdk.integrations.fastapi import (
    FastAPIInstrumentationMiddleware,
    HttpInstrumentationConfig,
)

app.add_middleware(
    FastAPIInstrumentationMiddleware,
    instrumentation=instrumentation,
    config=HttpInstrumentationConfig(
        include_paths=frozenset({"/v1/agent/run"}),
        include_methods=frozenset({"POST"}),
        exclude_paths=frozenset({"/health"}),
    ),
)

Matching uses exact decoded ASGI paths and uppercase methods; exclusions win. With no inclusion filters, every HTTP request not explicitly excluded is instrumented. Lifespan and WebSocket scopes pass through unchanged.

Each included request owns one async agent.run, and downstream explicit SDK spans inherit its trace and parent context. The root records only method, response status, response completion, disconnect observation, and application completion. Raw paths, route parameters, query strings, headers, cookies, client/server addresses, and request/response bodies are never recorded by the middleware. Paths are inspected only for local exact matching. Standard HTTP methods are recorded by name; every nonstandard method is emitted as the fixed OTHER sentinel rather than copying untrusted method text.

The root remains active until the ASGI application returns, including streamed response completion and response background work. Handled 5xx responses, incomplete responses, target exceptions, and cancellation produce an error root without fabricating exception content. Runtime instrumentation and collector failures remain fail-open and do not replace the target response or exception. Active cancellation finalizes collection without an awaited flush checkpoint and re-propagates the original cancellation object.

LangGraph wrapper

Install the optional framework extra:

uv add "accugents-sdk[langgraph]"

Wrap one compiled graph with the shared Instrumentation instance:

from accugent_sdk import AgentRunOptions, PromptVersion
from accugent_sdk.integrations.langgraph import LangGraphInstrumentation

instrumented_graph = LangGraphInstrumentation(
    graph=compiled_graph,
    instrumentation=instrumentation,
    options=AgentRunOptions(prompt_version=PromptVersion("prompt-v2")),
)

sync_result = instrumented_graph.invoke({"question": "What changed?"})
async_result = await instrumented_graph.ainvoke(
    {"question": "What changed?"},
)

Each invoke() or ainvoke() creates one agent.run when no run is active. When a caller such as the FastAPI middleware already owns the Agent run, scope entry fails open and the graph reuses that active trace context. Existing explicit SDK spans started inside graph nodes inherit the resulting root's trace and parent context. The wrapper forwards graph input and optional LangGraph RunnableConfig unchanged. Target exceptions, cancellation, and collector or instrumentation setup failures retain the SDK's fail-open behavior.

The MVP wrapper does not automatically instrument nodes or install LangGraph callbacks. Instrument LLM, tool, retrieval, and citation operations explicitly with the normal SDK child-span APIs. Only final-result invoke() and ainvoke() are wrapped in the MVP; streaming and batch methods remain outside this adapter. Run the complete sync/async example from the Backend repository root:

uv run --project packages/accugent-sdk --extra langgraph \
  python packages/accugent-sdk/examples/langgraph_wrapper.py

Instrumentation APIs

Configure one Instrumentation instance for an Agent identity and existing collector. Each Agent-run invocation generates a fresh run ID, trace ID, root span ID, and isolated sequence state.

from accugent_sdk import (
    AgentIdentity,
    AgentRunOptions,
    Instrumentation,
    InstrumentationConfig,
    PromptVersion,
    RetrievalDocumentMetadata,
    RetrievalMetadata,
    SpanKind,
    SpanOptions,
)

instrumentation = Instrumentation(
    sink=collector,
    config=InstrumentationConfig(
        identity=AgentIdentity(
            tenant_id="tenant-demo",
            project_id="project-demo",
            agent_id="support-agent",
            agent_version="v1",
        ),
    ),
)


@instrumentation.trace_async_agent_run(
    AgentRunOptions(prompt_version=PromptVersion("prompt-v2")),
)
async def answer() -> str:
    with instrumentation.start_span(
        SpanOptions(
            kind=SpanKind.RETRIEVAL,
            metadata=RetrievalMetadata(
                query="customer refund policy",
                top_k=1,
                documents=(
                    RetrievalDocumentMetadata(
                        document_id="policy-2026",
                        score=0.94,
                    ),
                ),
            ),
        ),
    ) as retrieval:
        retrieval.set_attribute("cache.hit", False)
    return "Grounded answer"

Sync entry points use trace_agent_run() and child operations use trace_span() or trace_async_span(). The equivalent start_agent_run(), start_async_agent_run(), and start_span() context managers expose a SpanHandle for terminal attributes and opt-in captured content.

Nested spans inherit the active trace and parent span through ContextVar state. Concurrent Agent runs receive separate run/trace identities and sequence counters. Exceptions keep their original type and identity while emitting an error span containing only error.type by default. Captured error details, prompts, responses, tool material, retrieval content, and citations remain opt-in and pass through the configured redaction policy before submission.

LlmCallMetadata, ToolCallMetadata, RetrievalMetadata, CitationMetadata, and ErrorMetadata require the Product trace specification's machine-consumed fields for canonical child spans. Missing or mismatched metadata suppresses that span and records an instrumentation failure on the root without replacing the Target Agent result or exception. Instrumentation derives latency_ms, status, and LLM total tokens. Reserved canonical keys cannot be injected through arbitrary attributes or overwritten through SpanHandle.set_attribute(). prompt.version is emitted only when AgentRunOptions.prompt_version is provided.

Synchronous Agent-run exit records a non-blocking flush request and therefore marks completeness as partial until a later boundary confirms delivery. Asynchronous Agent-run exit awaits the operational-event flush, emits root completeness evidence, and performs a second bounded flush for the terminal root span. Root attributes record root start/finish, operational-event flush outcome, dropped-event count, collector failures, instrumentation failures, and trace.collection.completeness. Global trace.completeness remains partial until the Backend combines collection evidence with terminal-root receipt and evaluator completion.

Queue pressure, delivery failure, sink submission or flush exceptions, and a dropped span do not replace the Target Agent's result or exception. Span scopes are single-use, closed parent contexts cannot emit late children, and SpanHandle inspection returns deep snapshots that cannot mutate queued material.

Capture and redaction

Content is excluded unless its category is enabled. Redaction returns a new value and never mutates the caller-owned content.

from accugent_sdk import (
    CapturePolicy,
    CapturedContent,
    RedactionConfig,
    sanitize_content,
)

result = sanitize_content(
    CapturedContent(
        response="Agent response",
        tool_arguments={"authorization": "Bearer secret"},
    ),
    policy=CapturePolicy(responses=True, tool_arguments=True),
    config=RedactionConfig(),
)

Collector lifecycle

The caller owns the AnyIO task group, so collector work cannot outlive its structured-concurrency scope. submit() uses a non-blocking bounded send and returns SubmitResult.DROPPED instead of waiting for queue capacity.

import anyio

from accugent_sdk import AsyncCollector, CollectorConfig

collector = AsyncCollector(
    transport=transport,
    config=CollectorConfig(queue_capacity=256, batch_size=32),
)

async with anyio.create_task_group() as task_group:
    await task_group.start(collector.run)
    collector.submit(span)
    await collector.aclose()

Expected TransportError and per-attempt timeouts are contained by the collector. Exhausted delivery records a collector failure and does not replace or raise through the Target Agent's response path.

Ingestion wire format

IngestionBatch projects a collector EventBatch into the signed HTTP request body without changing its batch identity or event order.

from datetime import datetime, timezone

from accugent_sdk import (
    IngestionBatch,
    IngestionNonce,
    canonical_ingestion_bytes,
)

wire_batch = IngestionBatch.from_event_batch(
    batch,
    sent_at=datetime.now(timezone.utc),
    nonce=IngestionNonce("request_nonce_1234"),
)
request_body = canonical_ingestion_bytes(wire_batch)

Canonical bytes are compact UTF-8 JSON with recursively sorted keys. SignedHttpTransport generates each request timestamp and nonce, hashes these exact bytes, and attaches HMAC headers. The wire-format layer itself performs no retry or network I/O.

Ingestion request signing

SdkWriteCredential keeps raw write-key material inside Pydantic SecretBytes. Signing copies no credential material into the resulting request value.

from accugent_sdk import (
    IngestionRequestTarget,
    SdkWriteCredential,
    sign_ingestion_batch,
    verify_ingestion_signature,
)

credential = SdkWriteCredential.model_validate(
    {
        "key_id": "sdk-key-2026-01",
        "secret": secret_bytes,
    },
)
target = IngestionRequestTarget(
    method="POST",
    scheme="https",
    authority="ingest.example",
    path="/api/v1/ingestion/batches",
)
signed_request = sign_ingestion_batch(
    wire_batch,
    credential,
    target=target,
)
assert verify_ingestion_signature(signed_request, credential)

The v2 signing input is newline-delimited in this exact order:

accugent.ingestion.batch.v2
POST
https
authority
path
key_id
timestamp
nonce
batch_id
sha256(canonical_request_body)

The signature value is v2=<hmac-sha256-hex>. Version 1 signatures are not accepted because they did not bind the request destination. Verification recomputes the body digest and uses constant-time comparison for key identity, body digest, and signature. The Backend persists encrypted credentials, supports provisioning and revocation through internal service functions, serializes revocation against acceptance, and enforces the replay window. Trusted provisioning, rotation, and revocation use the Backend's non-HTTP credential operator documented in docs/INGESTION_CONTRACT.md.

Signed HTTP transport

SignedHttpTransport implements the Collector EventTransport protocol. Each send attempt creates fresh UTC request metadata, projects and signs the retry-stable batch, and POSTs the exact signed bytes. Endpoints must use HTTPS and cannot contain user information, a query, or a fragment. Redirects are never followed, including when the caller injects its own redirect-enabled HTTP client.

from accugent_sdk import (
    HttpTransportConfig,
    SignedHttpTransport,
)

transport = SignedHttpTransport(
    config=HttpTransportConfig.model_validate(
        {
            "endpoint": ("https://accugent.example/api/v1/ingestion/batches"),
        },
    ),
    credential=credential,
)

try:
    await transport.send(batch)
finally:
    await transport.aclose()

Requests carry:

Content-Type
X-AccuGent-Key-Id
X-AccuGent-Timestamp
X-AccuGent-Nonce
X-AccuGent-Signature
X-AccuGent-Body-SHA256
Idempotency-Key

The optimized client enables HTTP/2, Brotli and Zstandard decoding, tuned connection pooling, split timeouts, TCP_NODELAY, sanitized request/response observability, and three connect-level retries. Redirect following remains disabled. HTTP status retry remains owned by AsyncCollector, so the transport performs one application-level request attempt per send() call.

Expected timeout, network, protocol, authentication, rate-limit, client, and server failures are mapped to stable TransportError.code values without including response bodies, raw client errors, request bodies, or credential material.

Backend ingestion boundary

The registered FastAPI endpoint streams at most 1 MiB, parses the signed headers above, authenticates the raw bytes before JSON parsing, requires the exact canonical request body, verifies timestamp, nonce, idempotency identity, and credential tenant/project scope, then delegates only an authenticated batch to an injected handler.

The endpoint returns 503 when SDK_CREDENTIAL_ENCRYPTION_KEY is absent, which disables production ingestion. It never acknowledges and drops a verified batch. The exact HTTP and security contract is documented in docs/INGESTION_CONTRACT.md.