import PythonSdkVersionNote from '../../_snippets/python-sdk-version-note.mdx'

The Python SDK makes it easy to capture events, evaluate feature flags, track errors, and more in your Python apps.

These docs cover version 7.x of the Python SDK, which requires Python 3.10 or higher. On Python 3.9? See supported versions.

import AgentIntegrationSection from "../../components/AgentIntegrationSection.mdx"

import PythonInstall from '../../integrate/_snippets/install-python.mdx'

The Python SDK includes an asyncio-native client in version 7.45.0 and later. Continue to use Posthog in synchronous apps. For an asyncio app, install the optional async dependencies:

pip install "posthog[async]>=7.45.0"

Import AsyncPosthog, the customer-facing name for AsyncClient. Both names provide the same async context manager and lifecycle methods. Keep one client for the lifetime of your app. For example, use a FastAPI lifespan handler:

import os
from contextlib import asynccontextmanager

from fastapi import FastAPI
from posthog import AsyncPosthog

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with AsyncPosthog(
        os.environ["POSTHOG_PROJECT_TOKEN"],
        host=os.environ["POSTHOG_HOST"],
        secret_key=os.environ.get("POSTHOG_FEATURE_FLAGS_SECURE_API_KEY"),
    ) as posthog:
        app.state.posthog = posthog
        yield

app = FastAPI(lifespan=lifespan)

Exiting the context calls shutdown(). This flushes buffered events, waits for in-flight operations, stops the workers, and closes the HTTP transport. If you don't use the context manager, call await posthog.shutdown() during app shutdown. await posthog.join() has the same effect.

capture() queues an event and returns without waiting for a network request. Don't await it:

posthog.capture(
    "event_name",
    distinct_id="user-distinct-id",
    properties={"source": "fastapi"},
)

Use capture_immediate() when your code must wait for that event's delivery attempt:

capture_id = await posthog.capture_immediate(
    "event_name",
    distinct_id="user-distinct-id",
)

Await evaluate_flags() once, then use its snapshot with synchronous in-memory accessors. Pass the same snapshot to capture() to attach the exact values used for branching without another feature flag request:

flags = await posthog.evaluate_flags("user-distinct-id")

if flags.is_enabled("new-checkout"):
    # Show the new checkout
    pass

posthog.capture(
    "checkout started",
    distinct_id="user-distinct-id",
    flags=flags,
)

The snapshot provides synchronous is_enabled(), get_flag(), and get_flag_payload() accessors. The awaited evaluate_flags() call also accepts groups, person_properties, group_properties, disable_geoip, flag_keys, and device_id arguments.

Initialize the client with a server-side secret API key as secret_key, then await the remote config request:

config = await posthog.get_remote_config_payload("landing-page-config")

See Remote config for setup and security details.

import IdentifyBackendCalloutPython from '../../_snippets/identify-backend-callout-python.mdx'

import PythonSendEvents from '../../integrate/send-events/_snippets/send-events-python.mdx'

The Python SDK captures identified events if the current context is identified or if you pass a distinct ID explicitly. These create person profiles. To set person properties in these profiles, include them when capturing an event:

# Passing a distinct id explicitly
posthog.capture(
    'event_name',
    distinct_id='user-distinct-id',
    properties={
        '$set': {'name': 'Max Hedgehog'},
        '$set_once': {'initial_url': '/blog'}
    }
)

# Using contexts
from posthog import new_context, identify_context
with new_context():
    identify_context('user-distinct-id')
    posthog.capture('event_name')

For more details on the difference between $set and $set_once, see our person properties docs.

To capture anonymous events without person profiles, set the event's $process_person_profile property to False. Events captured with no context or explicit distinct_id are marked as personless, and will have an auto-generated distinct_id:

posthog.capture(
    event='event_name',
    properties={
        '$process_person_profile': False
    }
)

Sometimes, you want to assign multiple distinct IDs to a single user. This is helpful when your primary distinct ID is inaccessible. For example, if a distinct ID used on the frontend is not available in your backend.

In this case, you can use alias to assign another distinct ID to the same user.

posthog.alias(previous_id='distinct_id', distinct_id='alias_id')

We strongly recommend reading our docs on alias to best understand how to correctly use this method.

import Contexts from './_snippets/contexts.mdx'

Group analytics allows you to associate an event with a group (e.g. teams, organizations, etc.). Read the Group Analytics guide for more information.

Note: This is a paid feature and is not available on the open-source or free cloud plan. Learn more on our pricing page.

To capture an event and associate it with a group:

posthog.capture('some_event', groups={'company': 'company_id_in_your_db'})

To update properties on a group:

posthog.group_identify('company', 'company_id_in_your_db', {
    'name': 'Awesome Inc.',
    'employees': 11
})

The name is a special property which is used in the PostHog UI for the name of the group. If you don't specify a name property, the group ID will be used instead.

The examples in this section use the synchronous Posthog client. For AsyncPosthog, use the awaited feature flag example. The returned snapshot uses the same accessors.

import FeatureFlagsLibsIntro from "../_snippets/feature-flags-libs-intro.mdx"

import PythonFeatureFlagsCode from '../../integrate/feature-flags-code/_snippets/feature-flags-code-python.mdx'

import LocalEvaluationIntro from "../../feature-flags/snippets/local-evaluation-intro.mdx"

For details on how to implement local evaluation, see our local evaluation guide.

In multi-worker or edge environments, you can implement custom caching for flag definitions using Redis, Cloudflare KV, or other storage backends. This enables sharing definitions across workers and coordinating fetches. See our guide for local evaluation in distributed environments for details.

Since experiments use feature flags, the code for running an experiment is very similar to the feature flags code. This example uses the synchronous Posthog client:

flags = posthog.evaluate_flags("user_distinct_id")
variant = flags.get_flag("experiment-feature-flag-key")

if variant == "variant-name":
    # Do something

With AsyncPosthog, await the evaluation: flags = await posthog.evaluate_flags("user_distinct_id"). The remaining snapshot access is the same.

It's also possible to run experiments without using feature flags.

Our Python SDK includes a built-in AI Observability feature. It enables you to capture LLM usage, performance, and more. Check out our analytics docs for more details on setting it up.

You can autocapture exceptions by setting the enable_exception_autocapture argument to True when initializing the PostHog client.

from posthog import Posthog

posthog = Posthog("<ph_project_token>", enable_exception_autocapture=True, ...)

You can also manually capture exceptions using the capture_exception method:

posthog.capture_exception(e, distinct_id='user_distinct_id', properties=additional_properties)

Contexts automatically capture exceptions thrown inside them, unless disable it by passing capture_exceptions=False to new_context().

The Python SDK can automatically capture the state of local variables when an exception occurs. This gives you a debugger-like view of your application state at the time of the error:

posthog = Posthog(
    "<ph_project_token>",
    enable_exception_autocapture=True,
    capture_exception_code_variables=True,
)

You can configure which variables are captured, masked, or ignored. See the code variables documentation for detailed configuration options.

Requires posthog version 7.58.0 or later.

Tracing is new in the Python SDK and its API can still change in a minor release. Spans you send are kept – it's the SDK surface that isn't frozen yet.

Tracing records spans – timed units of work – so you can see where time went in a request and how work fans out across your services. Spans created inside a context automatically carry the person and session they belong to, so a slow trace links back to the person who experienced it.

Tracing is off until you set the traces option. No OpenTelemetry dependency is required. For what you can do with spans once they arrive, see Distributed tracing.

from posthog import Posthog

posthog = Posthog(
    "<ph_project_token>",
    host="<ph_client_api_host>",
    traces={"service_name": "checkout-api"},
)

If you use the module-level API instead, set posthog.traces = {"service_name": "checkout-api"} alongside your other options, before you start the first span.

Set service_name – PostHog groups operations by service and span name.

Tracing is available on the synchronous Posthog client and the module-level API. AsyncPosthog doesn't support it yet.

start_span returns a span. Use it in a with block to make it the active span for the block and end it when the block exits. Spans started inside the block nest underneath it automatically.

with posthog.start_span("POST /checkout", kind="server") as span:
    span.set_attribute("plan", user.plan)

    with posthog.start_span("create-order"):
        order = create_order(cart)
    with posthog.start_span("charge-card"):
        stripe.charge(order)

If an exception escapes the block, the span records it, its status is set to error, and the exception propagates unchanged. KeyboardInterrupt, GeneratorExit, and asyncio.CancelledError still end the span but aren't recorded as failures.

The recorded exception includes the stack trace, which contains file paths from your server. If you'd rather those didn't leave your process, remove exception.stacktrace in before_span_send.

For work that can't wrap a block, call start_span without with. A span started this way isn't active, so spans started afterwards aren't its children unless you pass parent explicitly – and you must call end() yourself.

span = posthog.start_span("background-sync", attributes={"queue": "emails"})
try:
    # Explicitly parent a child to a span that isn't active.
    child = posthog.start_span("send-batch", parent=span)
    child.end()
finally:
    span.end()

get_active_span() returns the span active in the current context, or None when there isn't one.

The active span is tracked with contextvars, so it carries across await in asyncio code. Threads don't reliably inherit it – pass parent=span to continue the trace in a thread you start or a ThreadPoolExecutor task. A forked child process starts with no active span, so pass parent there too.

start_span always returns a usable span, even when tracing is off, so your code never needs to check whether tracing is enabled.

Span names should be low-cardinality operation names – GET /users/:id, not GET /users/123. Variable values belong in attributes. Strings, booleans, integers, and floats keep their type. Lists and dictionaries are sent as arrays and maps, but PostHog stores them as serialized strings. Setting an attribute to None removes it, and any other value is converted to a string.

with posthog.start_span("GET /users/:id", kind="server") as span:
    span.set_attributes({"user.id": user_id, "db.rows": len(rows)})
    span.add_event("cache-miss")

    if not rows:
        span.set_status("error", "user not found")

Every method except traceparent(), tracestate(), and end() returns the span, so calls chain. Calls after end() are ignored.

start_span takes these keyword arguments:

Spans use W3C Trace Context, so a trace can span several services. Pass an inbound traceparent header as parent to continue a trace another service started, and send span.traceparent() onward when you call out.

import requests
from flask import request


@app.post("/checkout")
def checkout():
    with posthog.start_span(
        "POST /checkout",
        kind="server",
        parent=request.headers.get("traceparent"),
    ) as span:
        traceparent = span.traceparent()

        requests.post(
            "https://payments.internal/charge",
            headers={"traceparent": traceparent} if traceparent else {},
        )

        return {"status": "ok"}

A malformed traceparent starts a new trace rather than raising. A missing one (None) falls back to the active span, if there is one.

A continued trace propagates the sampled flag it was handed, so a downstream sampler sees the decision the head service made. PostHog itself doesn't sample – a span is recorded and exported whichever way that flag is set.

Spans created inside a context that has a distinct ID or session ID automatically carry posthogDistinctId and sessionId attributes, which is what makes a trace reachable from a person or a Session Replay recording. In Django, the contexts middleware sets these for every request. Elsewhere, set them yourself:

from posthog import new_context, identify_context, set_context_session

with new_context():
    identify_context(user.id)
    set_context_session(session_id)

    with posthog.start_span("POST /checkout"):
        process_order()

Spans created outside a context with those values omit the attributes.

before_span_send runs on every finished span before it's queued for export. It receives the span as a dict with name, kind, status, attributes, events, start_time_ns, end_time_ns, trace_id, span_id, and parent_span_id. Edit it and return it, or return None to drop the span entirely.

def scrub_spans(span):
    if span["attributes"].get("http.route") == "/health":
        return None

    span["attributes"].pop("http.request.header.authorization", None)
    return span


posthog = Posthog(
    "<ph_project_token>",
    host="<ph_client_api_host>",
    traces={
        "service_name": "checkout-api",
        "before_span_send": scrub_spans,
    },
)

Attributes are plain Python values, not the OTLP wire encoding. The hook runs after PostHog attaches posthogDistinctId and sessionId, so those are visible to the hook and can be scrubbed too. An exception's stack trace is on its event, under event["attributes"]["exception.stacktrace"].

• trace_id, span_id, and parent_span_id are read-only. Rewriting them would orphan child spans that have already been exported, so changes are reverted.
• A hook that raises drops the span rather than exporting it without scrubbing.
• Pass a list to run several hooks in order. The first one to return None stops the chain.
• The hook must be a regular function. An async hook drops every span.
• If an entry isn't callable, tracing turns off for the client rather than exporting spans the hook was meant to scrub.

A span is capped at 128 attributes and 128 events, each event at 128 attributes, and each string attribute value at 8192 characters. The endpoint rejects a span that's too large, and a rejected span is lost whole rather than truncated, so the caps bound a span before it gets there.

Past the cap, the earliest attributes and events are kept and the number dropped is reported alongside the span, so a truncated span reads as truncated rather than as quietly incomplete. The attributes PostHog attaches itself – posthogDistinctId and sessionId – don't count toward the cap and are never dropped, so a span at the limit still links back to its person and session.

The event cap is absolute: an exception event the SDK records for you spends an ordinary slot like any other. A span that fills its events and then raises keeps its error status but not the exception detail, and reports the loss as a dropped event. Raise max_events_per_span on spans that record many events and can also fail.

The length bound reaches inside a value, including strings nested in lists and dictionaries, and applies to exception.stacktrace like any other attribute – a long stack trace keeps its last 8192 characters. All four caps are re-applied after before_span_send, so a hook that enriches a span can't push it back over.

An invalid value falls back to its default with a warning. The exception is before_span_send: an entry that isn't callable turns tracing off.

Spans are exported on a background interval, even with sync_mode on. Both flush() and shutdown() export spans that have already ended. A span still open at flush() is exported once it ends; a span still open at shutdown() is discarded with a warning, so end your spans before shutting down – a with block does this for you. shutdown() gives queued spans up to 30 seconds to send.

If your serverless handler reuses a client across invocations, call flush() before returning. It flushes events and ended spans concurrently without stopping the client. Reserve shutdown() for a disposable client or process exit.

def handler(event, context):
    try:
        with posthog.start_span("handler"):
            do_work()
    finally:
        posthog.flush()

A script that exits without calling shutdown() still gets a brief best-effort flush at exit, but don't rely on it for spans you need.

Before posthog-python v3.0, we added GeoIP properties to all incoming events by default. We also used these properties for feature flag evaluation, based on the IP address of the request. This isn't ideal since they are created based on your server IP address, rather than the user's, leading to incorrect location resolution.

As of posthog-python v3.0, the default now is to disregard the server IP, not add the GeoIP properties, and not use the values for feature flag evaluations.

You can go back to previous behavior by doing setting the disable_geoip argument in your initialization to False:

posthog = Posthog('api_key', disable_geoip=False)

The list of properties that this overrides:

1. $geoip_city_name
2. $geoip_country_name
3. $geoip_country_code
4. $geoip_continent_name
5. $geoip_continent_code
6. $geoip_postal_code
7. $geoip_time_zone
8. $geoip_subdivision_1_code
9. $geoip_subdivision_1_name
10. $geoip_subdivision_2_code
11. $geoip_subdivision_2_name

You can also explicitly chose to enable or disable GeoIP for a single capture request like so:

posthog.capture('test_event', disable_geoip=True|False)

If you're not seeing the expected events being captured, the feature flags being evaluated, or the surveys being shown, you can enable debug mode to see what's happening.

You can enable debug mode by setting the debug option to True in the PostHog object. This will enable verbose logs about the inner workings of the SDK.

posthog.debug = True # +

You can disable requests during tests by setting the disabled option to True in the PostHog object. This means no events will be captured or no requests will be sent to PostHog.

if settings.TEST:
    posthog.disabled = True

The SDK uses HTTP connection pooling internally for better performance. These settings typically need not be changed, but in some environments, such as when running behind NAT gateways, pooled connections may be terminated non-gracefully, causing request failures.

You can configure connection behavior in several ways. The following settings should be called during initialization, before any API requests are made.

TCP keepalive probes help prevent idle connections from being dropped by network infrastructure. This is the recommended approach for most cases where idle connections are terminated.

import posthog

posthog.enable_keep_alive()

This enables TCP keepalive with sensible defaults (60 second idle time, 60 second probe interval, 3 probes before timeout).

If you need each request to use a fresh connection, you can disable connection reuse entirely. This will incur additional overhead per request but may be desirable in some circumstances.

import posthog

posthog.disable_connection_reuse()

For advanced use cases, you can configure arbitrary socket options on the underlying HTTP connection.

import socket
import posthog

posthog.set_socket_options([
    (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
    # Add additional socket options as needed
])

Pass None to set_socket_options() to reset to default behavior.

Use before_send to modify or drop events before they are queued for delivery. Return the modified event dictionary to send it, or None to drop it.

from typing import Any

import posthog


def scrub_pii(event: dict[str, Any]) -> dict[str, Any] | None:
    properties = event.get("properties", {})

    if "email" in properties:
        email = properties["email"]
        properties["email"] = f"***@{email.split('@', 1)[1]}" if "@" in email else "***"

    if event.get("event") == "test_event":
        return None

    return event


client = posthog.Client(
    "<ph_project_api_key>",
    before_send=scrub_pii,
)

If your callback raises an exception, the SDK logs the error and continues with the original unmodified event.

You can use the Python or Node SDK to run historical migrations of data into PostHog. To do so, set the historical_migration option to true when initializing the client.

import SDKMigration from "../../migrate/_snippets/sdk-migration.mdx"

By default, the synchronous Posthog client buffers events before sending them to the capture endpoint. A serverless platform can freeze or terminate the runtime before background delivery finishes.

• For a client shared across invocations, call posthog.flush() before the handler returns, including when application code throws. This blocking call attempts to deliver queued events without stopping the client. Its timeout_seconds argument defaults to 10 seconds.
• For a disposable client or process exit, call posthog.shutdown() after its final work. It attempts to deliver queued events and cleans up the client. Don't reuse the client afterward. Unlike Node's shutdown(), Python's shutdown() takes no timeout argument.
• Alternatively, enable sync_mode when initializing the client so each posthog.capture() call attempts delivery before it returns.

These calls do not guarantee that PostHog received every event. If you use distributed tracing, sync_mode doesn't apply to spans. For a shared client, call posthog.flush() before the handler returns – see Shutdown and short-lived processes.

Keep one AsyncPosthog client for the lifetime of your application. Use buffered capture() by default, or await capture_immediate() when one invocation must wait for an event's delivery attempt. Call await posthog.shutdown() once during application cleanup. Don't shut down the client after each request.

See our Django docs for how to set up PostHog in Django. The Django contexts middleware adds request properties and authenticated user identity to events captured during a request. With SDK 7.71.0 or later, it also reads browser session and distinct ID headers by default. Set POSTHOG_MW_CAPTURE_TRACING_HEADERS = False to disable header extraction.

Version 7.69.0 and later of the Python SDK includes framework-independent ASGI middleware for FastAPI, Starlette, Litestar, and raw ASGI applications. It creates a PostHog context for each HTTP or WebSocket request, enriches captured events with request properties, and captures exceptions that escape the application.

For FastAPI or Starlette, add the middleware to your app:

from fastapi import FastAPI
from posthog import Posthog
from posthog.integrations.asgi import PosthogASGIMiddleware

posthog = Posthog(
    "<ph_project_token>",
    host="<ph_client_api_host>",
    enable_exception_autocapture=True,
)

app = FastAPI()
app.add_middleware(PosthogASGIMiddleware, client=posthog)

The middleware's capture_exceptions option defaults to None, which inherits the client's enable_exception_autocapture setting. Set it to True or False to override the client. Use request_filter to skip selected requests and extra_properties to add properties from the ASGI scope. Both callbacks can be synchronous or asynchronous.

With SDK 7.71.0 or later, the middleware reads X-PostHog-Distinct-ID and X-PostHog-Session-ID by default. Set capture_tracing_headers=False to disable header extraction. Clients can control these headers. Use them only as analytics context, never for authentication or authorization.

To instrument another ASGI framework or a raw ASGI app, wrap the application directly:

app = PosthogASGIMiddleware(app, client=posthog)

As our open source project PostHog shares the same module name, we created a special posthoganalytics package, mostly for internal use to avoid module collision. It is the exact same.

This library is largely based on the analytics-python package.