Python

The Python SDK supports plain Python plus Django, Flask, FastAPI, Celery, grpc.aio, standard-library logging, and opt-in HTTP client and LLM integrations.

Install and start

After the registry release:

pip install anectico
# Or include an integration extra:
# anectico[django], anectico[flask], anectico[fastapi], anectico[celery], anectico[grpc]

During early access, use the wheel supplied during onboarding or install the source package:

pip install "anectico @ git+https://github.com/anectico/anectico.git@main#subdirectory=sdks/python"

import os
import anectico

client = anectico.AnecticoClient(
    api_key=os.environ['ANECTICO_API_KEY'],
    service_name='orders-api',
)
client.start()

try:
    process_order()
except Exception as exc:
    client.capture_error(exc, tags={'component': 'checkout'})
    raise

Keep one client for the process and call client.stop() during graceful shutdown. Inspect its ShutdownResult when delivery matters: success is True only if every enabled provider flushed and shut down, False on a provider failure or aggregate deadline, and None when the client was already stopped. A context manager can start and stop the client for scripts that do not need to inspect that result.

Use a project-scoped key with ingest:write. Add analytics:write when calling identify, group, or diagnostic-event methods, as this guide does. Do not give an application key read or management scopes.

Keep an existing OpenTelemetry setup

Use existing-provider mode when the process already owns OpenTelemetry resources, sampling, batching, exporters or a Collector, instrumentation, and provider lifecycle. Add Anectico’s passive identity processors while building those providers, register the providers, and only then start Anectico:

from opentelemetry import _logs as otel_logs
from opentelemetry import metrics, trace
import anectico

# Keep your existing Resource, sampler, readers, exporters, and instrumentation.
tracer_provider.add_span_processor(anectico.BaggageIdentitySpanProcessor())
tracer_provider.add_span_processor(existing_span_export_processor)
logger_provider.add_log_record_processor(
    anectico.BaggageIdentityLogRecordProcessor()
)
logger_provider.add_log_record_processor(existing_log_export_processor)

trace.set_tracer_provider(tracer_provider)
metrics.set_meter_provider(meter_provider)
otel_logs.set_logger_provider(logger_provider)

client = anectico.AnecticoClient(
    api_key=os.environ["ANECTICO_API_KEY"],
    service_name="orders-api",
    open_telemetry_mode="existing",
)
client.start()

You can set ANECTICO_OTEL_MODE=existing instead. Anectico then creates no providers, exporters, export processors, metric readers, propagator, HTTP client instrumentation, or standard-library log bridge. Its span/error/AI/tool/agent and metric helpers use the application globals. log_event() uses the global Logger Provider and safely does nothing when no real one is registered.

The identity processors must precede the application’s export processors so spans and logs reach the exporter with anectico.distinct_id. They have no exporter or lifecycle behavior. Do not put person identity on metrics. Anectico’s managed resource, sampling, OTLP endpoint, batching, propagation, and logging options do not reconfigure application providers; the application sampler decides which helper spans export. Signal enable flags still gate helper emission.

Existing mode never flushes or shuts down application providers: client.flush() returns False, and client.stop() reports attempted=False, success=None. After the final helper call, stop Anectico and run the application’s existing provider flush/shutdown hook. To roll back only the Anectico layer, remove client startup and both identity processors. Restore the old exporter or Collector destination separately when rolling back OTLP delivery itself.

Add framework middleware

FastAPI:

from fastapi import FastAPI
from anectico import AnecticoClient
from anectico.integrations.fastapi import AnecticoFastAPIMiddleware

app = FastAPI()
client = AnecticoClient(api_key=os.environ['ANECTICO_API_KEY'], service_name='orders-api')
client.start()

async def authenticated_customer_id(scope):
    principal = await verify_bearer_token(scope)  # your server-side auth
    return principal.customer_id

app.add_middleware(
    AnecticoFastAPIMiddleware,
    client=client,
    request_identity_resolver=authenticated_customer_id,
)

The resolver runs before the server span starts and may be synchronous or asynchronous. Return only an identity established by server-side authentication; never copy a public header, query parameter, or caller-supplied baggage directly. The resolved identity is async-local, so interleaved requests, child spans, logs, errors, and trusted outbound calls stay with the correct customer.

Matched requests are named from the bounded FastAPI route template, such as POST /loans/{applicant_id}. Anectico exports the URL origin plus that template, never the concrete path, query string, or fragment; unmatched requests remain method-only and origin-only. Cancelled ASGI requests close with anectico.request.outcome=cancelled and a measured duration without creating an Issue. Pass skip_paths=set() when you intentionally want to instrument every path, including the normally excluded health/docs routes.

Flask uses AnecticoFlaskMiddleware(app, client). Register it after constructing the application and its routes; the middleware preserves Flask responses and error handlers, names matched requests from their bounded Werkzeug rule, and exports only the URL origin plus that rule. Concrete path identifiers, query values, and fragments are never exported; unmatched requests are named with the HTTP method only. Unhandled exceptions are captured before Flask writes its framework error log, so the diagnostic log and canonical Issue stay linked and the request span remains open through error handling.

from anectico.integrations.flask import AnecticoFlaskMiddleware

AnecticoFlaskMiddleware(app, client)

Outbound requests/httpx instrumentation remains opt-in. When a Flask route calls a trusted internal service, install the matching HTTP extra and allow only that service’s URL prefix; request headers and bodies are not captured by Anectico’s default configuration.

client = AnecticoClient(
    api_key=os.environ['ANECTICO_API_KEY'],
    service_name='legacy-policy-api',
    instrument_http_clients=True,
    propagate_trace_header_urls=('https://rating.internal',),
)

For Gunicorn, stop the process-local programmatic client from worker_exit so its final trace, log, and metric batches flush:

# gunicorn.conf.py
def worker_exit(server, worker):
    from insurance.app import client
    outcome = client.stop()
    if outcome.success is False:
        server.log.error('Anectico telemetry shutdown failed: %s', outcome.errors)

Create AnecticoCeleryIntegration(app, client) in every Celery producer and worker process. It injects W3C trace/person context into private broker headers and activates that context for each worker attempt, so retries, task children, logs, and terminal error capture stay on one causal trace and customer. Anectico does not copy task arguments, keyword arguments, return values, or retry exception messages into Celery span attributes. Do not accept broker messages from untrusted publishers.

Celery’s default worker logging setup may replace the root logger after Django or the application starts. The integration restores Anectico’s log bridge from Celery’s post-setup signals, so worker logs remain exported without disabling worker_hijack_root_logger.

Flush each prefork worker’s process-local client at shutdown:

import logging
from celery.signals import worker_process_shutdown

worker_logger = logging.getLogger(__name__)

@worker_process_shutdown.connect(weak=False)
def flush_anectico_worker(**_kwargs):
    if client.is_running:
        outcome = client.stop()
        if outcome.success is False:
            worker_logger.error(
                'Anectico telemetry shutdown failed: %s',
                outcome.errors,
            )

AnecticoCeleryIntegration.close() disconnects only its Celery signal receivers; the application still owns AnecticoClient.stop().

For Django, place Anectico after session and authentication middleware so the buyer is available before the view runs:

# settings.py
MIDDLEWARE = [
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'anectico.integrations.django.AnecticoDjangoMiddleware',
]

ANECTICO = {
    'API_KEY': os.environ['ANECTICO_API_KEY'],
    'SERVICE_NAME': 'marketplace-web',
    'SERVICE_VERSION': os.environ.get('APP_RELEASE', 'unknown'),
    'ENVIRONMENT': os.environ.get('ANECTICO_ENVIRONMENT', 'production'),
    'ENDPOINT': os.environ.get('ANECTICO_ENDPOINT', 'https://api.anectico.com'),
    # Optional callable or dotted path. The default uses authenticated user.pk.
    'USER_ID_RESOLVER': 'marketplace.observability.anectico_distinct_id',
}

# marketplace/observability.py
def anectico_distinct_id(request):
    user = getattr(request, 'user', None)
    if not user or not user.is_authenticated:
        return None
    return user.customer_id

The Django middleware creates one server span, privacy-safe ORM child spans, correlated standard logs, and one canonical capture for an unhandled view exception. It records the URL origin and path but never query values, fragments, usernames, or email addresses. Keep recognizable profile fields in sync_person(id, properties) at a trusted user creation/profile-update boundary; unlike identify, it does not change process-wide identity.

Gunicorn must flush each worker rather than only the master:

# gunicorn.conf.py
from anectico.integrations.django import shutdown_django_client

def worker_exit(server, worker):
    outcome = shutdown_django_client()
    if outcome.success is False:
        server.log.error('Anectico telemetry shutdown failed: %s', outcome.errors)

shutdown_django_client() returns the same delivery outcome and then detaches the process-local auto-created client. A repeated worker hook is harmless: attempted=False and success=None, rather than a false claim that another batch was flushed.

Framework middleware scopes identity and trace context to each request. Incoming identity baggage is ignored by default. Enable trust_incoming_identity=True only behind a gateway that strips and recreates untrusted baggage.

Connect trusted grpc.aio services

Install anectico[grpc], then add the client interceptor only to channels for trusted internal destinations. Add the server interceptor to each asynchronous gRPC service:

import grpc
from anectico.integrations.grpc import (
    AnecticoAioServerInterceptor,
    AnecticoAioUnaryUnaryClientInterceptor,
)

profile_channel = grpc.aio.insecure_channel(
    'dns:///profile.internal:50051',
    interceptors=[AnecticoAioUnaryUnaryClientInterceptor(client)],
)

grpc_server = grpc.aio.server(
    interceptors=[
        AnecticoAioServerInterceptor(client, trust_incoming_identity=True),
    ],
)

The unary-unary client span injects W3C traceparent, tracestate, and identity baggage into gRPC metadata; the server span extracts the same context, preserving one trace through FastAPI and each internal RPC. Propagation metadata and application messages are never copied into span attributes. The server rejects inbound identity by default. Enable trust_incoming_identity=True only on a private boundary that rejects or overwrites metadata from public callers. Typed non-OK gRPC results set rpc.grpc.status_code and an error span status without changing the application exception or retry behavior. These transport spans stay visible in traces and service health but do not create generic transport Issues. Call client.capture_error(...) once at the domain handling boundary when the failure should become a typed Issue, normally after retries are exhausted.

Scope identity safely

For a single-user process or an explicit login merge:

client.identify('user_8842', properties={'email': 'buyer@acme.example'})
client.group('company', 'acme', {'plan': 'enterprise'})
# On logout:
client.reset()

Reset clears the prior identity, groups, global error-user context, and uncaptured breadcrumbs. Telemetry captured before reset remains queued with its original attribution.

Do not switch process-wide identity per request in a concurrent server. Use middleware, or the async-safe shared identity context:

from anectico import reset_context_distinct_id, set_context_distinct_id

token = set_context_distinct_id(user_id)
try:
    handle_request()
finally:
    reset_context_distinct_id(token)

Record an application metric

client.record_metric(
    'checkout.queue_depth',
    queue.qsize(),
    labels={'region': 'eu-west', 'worker_pool': 'payments'},
)

record_metric records a gauge value. Use client.meter to create an explicit counter or histogram. Keep labels low-cardinality; do not add customer, order, request, trace, or session IDs.

Propagate only to trusted services

Install anectico[http] and, when needed, anectico[httpx]. Then explicitly allow the internal URL prefixes that may receive W3C trace context and customer identity:

client = AnecticoClient(
    api_key=os.environ['ANECTICO_API_KEY'],
    service_name='orders-api',
    instrument_http_clients=True,
    propagate_trace_header_urls=['https://billing.internal.example'],
)

The default allowlist is empty. Calls from uninstrumented clients remain unlinked.

Logs and AI calls

In managed mode, starting the client bridges standard-library logging at INFO and above by default, preserving the active trace and customer context:

import logging

logging.getLogger('orders').error(
    'payment failed',
    extra={
        'employee_id': 'employee-123',
        'payroll_period': '2026-07',
        'details': {
            'result': 'declined',
            'authorization': 'Bearer provider-secret',
        },
    },
)

If Django or another framework logs an exception that Anectico already captured, the log is linked to the canonical error occurrence and stays visible in Logs without creating a duplicate Issue. Independent error logs remain first-class Issues.

The logging bridge preserves safe extra fields and removes reserved LogRecord metadata. It recursively replaces built-in credential and financial-account fields with [REDACTED] before OTLP export; nested values are stored as bounded JSON when OpenTelemetry cannot represent them directly. Message and exception text also redact keyed credentials, Bearer/JWT values, credential URLs, IBANs, and valid payment-card-like values. Add application-specific field names through AnecticoClient(log_redaction_fields=[...]), AnecticoConfig(log_redaction_fields=(... ,)), or the comma-separated ANECTICO_LOG_REDACTION_FIELDS environment variable. Custom names extend the built-in protection; they never disable it.

Use anectico[openai] for synchronous, asynchronous, or streaming Chat Completions, and anectico[anthropic] for non-streaming messages. For an OpenAI stream, pass stream_options={"include_usage": True} and consume it to completion so Anectico can record the final token and cost fields. Closing early records one error-status call. Prompt and completion capture is opt-in and remains off by default.

Verify and recover

Generate one framework request for a known test customer. In Customers, confirm its trace, log, and test error appear together. If they do not, enable ANECTICO_DEBUG=true, verify the key and project, and inspect the result from client.stop().

The Python SDK rejects malformed or unsupported collector endpoints before starting. HTTP endpoints must be absolute http(s) URLs; gRPC endpoints must be an http(s) authority URL without a path or a host:port target. Signal-specific OTEL_EXPORTER_OTLP_*_ENDPOINT overrides are validated for enabled signals. HTTP certificate verification, TLS configuration, and client certificate failures are attempted once and reported with a sanitized diagnostic; verification stays enabled. Connection loss and retryable 408 or 5xx responses retain the same serialized batch through the configured export_timeout_ms window. Jittered exponential backoff is bounded by the remaining deadline, shutdown interrupts it, and exhaustion does not start a nested retry loop.

• Identify customers
• Capture errors and releases
• Record application metrics
• Record LLM calls
• Python SDK API reference
• Configuration reference