# Python

> Instrument Python services with metrics, telemetry, errors, and request-safe customer identity.

Canonical page: https://anectico.com/docs/instrument/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

During early access, install the wheel supplied during onboarding. Replace the path with its
actual filename:

```bash
pip install /path/to/anectico.whl
```

With authorized repository access, you can instead install the source package:

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

Pin the onboarding commit instead of `main` for reproducible builds. After the public PyPI release,
use `pip install anectico` or the integration extra you need: `anectico[django]`, `anectico[flask]`,
`anectico[fastapi]`, `anectico[celery]`, or `anectico[grpc]`.

```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:

```python
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:

```python
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.

```python
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.

```python
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:

```python
# 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:

```python
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:

```python
# 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',
}
```

```python
# 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:

```python
# 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:

```python
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:

```python
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:

```python
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

```python
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:

```python
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:

```python
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 synchronous, asynchronous, or streaming messages — a streamed call from
either wrapper records one span when it is exhausted, fails, or is abandoned. Stopping early
(`break`, an early `close()`, leaving the `with` block) is not an error: it records a non-error span
carrying `stream_abandoned` instead. 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; without
it, latency and status are still recorded but token/cost fields are unavailable. 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](/docs/instrument/identity)
- [Capture errors and releases](/docs/instrument/errors)
- [Record application metrics](/docs/instrument/metrics)
- [Record LLM calls](/docs/instrument/llm-calls)
- [Python SDK API reference](/docs/reference/python-sdk)
- [Configuration reference](/docs/reference/configuration)