Python SDK Reference

sdk-garden-python is a lightweight, stdlib-only observability client for Python. Embed it inside your published Python SDK to send initialization, method-call, and error telemetry to SDK Garden on a background daemon thread — your users will never experience a delay or exception from the instrumentation layer.

Package: sdk-garden-python
Requires: Python 3.8+, no third-party dependencies.

Installation

pip install sdk-garden-python

Configuration

All options are passed to the SdkGarden constructor.

Parameter	Type	Default	Required	Description
`api_key`	`str`	—	Yes	Your SDK Garden API key. Must start with `sgk_`.
`sdk_name`	`str`	—	Yes	The name of your SDK, e.g. `"my-payments-sdk"`.
`sdk_version`	`str`	—	Yes	The version of your SDK, e.g. `"1.2.3"`.
`endpoint`	`str`	`"https://ingest.sdkgarden.com/track"`	No	Override the ingest endpoint.
`debug`	`bool`	`False`	No	Print `[SDK Garden]` prefixed debug lines to stdout. The API key is never logged.
`disabled`	`bool`	`False`	No	When `True`, all methods become no-ops. No thread is spawned, no network calls made.
`batch_size`	`int`	`10`	No	Maximum events sent in a single HTTP POST, from 1 to 100. A full batch is sent without waiting for the next cycle.
`flush_interval`	`float`	`5.0`	No	Seconds between flush cycles on the background thread, from 0.1 to 86400.

API

`SdkGarden(api_key, sdk_name, sdk_version, **kwargs)`

Construct a new SDK Garden client instance. The constructor does not start the background thread — call .start() after construction.

Parameters: See Configuration above.

Returns: SdkGarden instance.

Example:

from sdk_garden import SdkGarden

sg = SdkGarden(
    api_key='sgk_live_abc123',
    sdk_name='acme-payments',
    sdk_version='2.1.0',
)
sg.start()

Or use method chaining:

sg = SdkGarden(
    api_key='sgk_live_abc123',
    sdk_name='acme-payments',
    sdk_version='2.1.0',
).start()

`.start()`

Spawn the background daemon thread and begin the flush loop. Returns self for method chaining.

Returns: SdkGarden (self)

Behavior:

When disabled=True, returns immediately without spawning a thread.
Idempotent — calling .start() again while the thread is alive is a no-op. After .shutdown() it does nothing.
The thread is a daemon thread, so it never keeps your process alive. When the interpreter exits, events still in the queue are sent first, for at most 2 seconds.
A child process created with os.fork() (for example a prefork server worker) starts with an empty queue and gets its own background thread on its first event. On older Pythons such as 3.8, don't track that first event from a signal handler: starting a thread there can deadlock. The parent sends what it had queued before the fork.

Example:

# Chained initialization (recommended)
sg = SdkGarden(api_key='...', sdk_name='...', sdk_version='...').start()

# Or separate construction and start
sg = SdkGarden(api_key='...', sdk_name='...', sdk_version='...')
# ... any setup ...
sg.start()

Automatically enqueues sdk_initialized with server-side environment metadata: runtime: "cpython" and runtime_version (e.g. "3.11.4"). No browser fields (screen dimensions, viewport, etc.) are sent — only what is meaningful in a server context.

Calling track() before start()

track() is safe to call before start(). Events are queued immediately and will be flushed once start() is called.

`.track(event_name, properties=None)`

Enqueue a named event. Use this for any event that isn't a method call or error.

Parameters:

Parameter	Type	Description
`event_name`	`str`	Name of the event, e.g. `"feature_used"` or `"config_loaded"`.
`properties`	`dict[str, Any] \| None`	Optional key/value metadata attached to the event.

Returns: None

Behavior:

Never blocks the caller.
Serializes the event right away, so later changes to properties are not sent. Values JSON cannot encode (datetime, Decimal, ...) are sent as strings. An event that cannot be serialized at all (a circular reference), holds nan or inf (not valid JSON) or is over 64 KiB serialized is dropped on its own, so it never costs the rest of its batch. For the same reason NUL characters and unpaired surrogates (say, from a surrogateescape-decoded path), which the server cannot store, are sent as U+FFFD, and an event whose name is not a string of at most 100 characters is dropped.
Silently discards events if the queue is full (max 100 events) or disabled=True.
Never raises an exception.

Example:

sg.track('feature_used', {'feature': 'streaming', 'model': 'gpt-4'})
sg.track('config_loaded', {'has_custom_endpoint': True})

# Custom event with no properties
sg.track('cache_cleared')

`.track_method(method_name)`

Return a decorator that wraps a function with method-call telemetry. The decorated function behaves identically to the original — same signature, same return value, same exceptions. An async def function stays a coroutine function: the call is timed until the coroutine finishes, and an exception it raises is recorded. For a generator function, the call is recorded when the generator is created.

Parameters:

Parameter	Type	Description
`method_name`	`str`	The name that will appear in the dashboard for this method.

Returns: A decorator (fn: F) -> F.

Emits:

Event	When
`method_called`	Every invocation, before the function runs. Includes `args_count`.
`method_success`	When the function returns (for `async def`, when the coroutine finishes). Includes `method`, `duration_ms`, and `is_async: true` for `async def`.
`sdk_error`	When the function raises. Includes `method`, `duration_ms`, `error_message`, `error_type`. The original exception is re-raised.

Example — module-level function:

@sg.track_method('process_payment')
def process_payment(amount: float, currency: str) -> dict:
    # your implementation
    return {'status': 'ok', 'amount': amount}

Example — class method:

class PaymentsClient:
    def __init__(self, sg: SdkGarden):
        self.sg = sg

    @property
    def _sg(self):
        return self.sg

    @sg.track_method('charge')
    def charge(self, amount: float) -> dict:
        ...

Example — wrap an existing function:

from mylib import generate_token

tracked_generate_token = sg.track_method('generate_token')(generate_token)

When disabled=True, the decorator returns the original function unwrapped.

`.flush(timeout=2.0)`

Ask the background thread to send every queued event now, and wait until it has. Useful at the end of a serverless handler, a short script or a test.

Parameters:

Parameter	Type	Description
`timeout`	`float`	Seconds to wait at most. Default `2.0`. Events not sent by then stay queued for the background thread. Since only the background thread sends, the timeout holds even while a request hangs.

Returns: None. Never raises. Does nothing before .start(), when disabled=True, or after .shutdown(). Safe to call from a signal handler.

def handler(event, context):
    result = do_work(event)
    sg.flush()
    return result

`.shutdown(timeout=2.0)`

Send every queued event, then stop the client for good: the background thread exits, and track, track_error and tracked functions do nothing. start() cannot restart a client that was shut down; create a new one instead.

Parameters:

Parameter	Type	Description
`timeout`	`float`	Seconds to wait at most. Default `2.0`. Events not sent by then are dropped. `0` stops the client without sending anything.

Returns: None. Never raises. Calling it again is harmless.

The client calls shutdown() for you when the interpreter exits, so you only need it to stop earlier, for example from your SDK's own close(). It is safe to call from a signal handler, such as a SIGTERM handler that closes your SDK, even when the signal lands in the middle of a tracking call.

`.track_error(error, context=None)`

Enqueue an sdk_error event for an exception you have already caught. The error is not re-raised — you control propagation.

Parameters:

Parameter	Type	Description
`error`	`Exception`	The exception to record. `str(error)` is PII-scrubbed before sending.
`context`	`dict[str, Any] \| None`	Optional additional metadata merged into the event properties.

Returns: None

PII scrubbing: Email addresses matching the standard user@example.com pattern are replaced with [REDACTED] in the error message before the event is enqueued. Only the start of the message is sent: its first 1000 characters, cut to 1000 UTF-8 bytes, as much as the server keeps. Apart from email addresses it is sent as it is, so whatever your SDK puts in an exception message (a hostname, a URL, an ID) is sent too.

Example:

try:
    connect_to_database(config)
except Exception as e:
    sg.track_error(e, {'phase': 'startup', 'config_has_tls': config.get('tls')})
    raise  # you decide whether to re-raise

Behavior guarantees

Silent network failure

All exceptions in the HTTP flush path are caught and discarded — including socket timeouts (2-second limit), DNS errors, and HTTP error responses. Your program will never see an exception from sdk-garden-python.

Daemon thread exit

The background flush thread is a daemon thread, so it never keeps your process alive, and it is the only thread that makes requests. When the interpreter exits normally, an atexit hook has it send the events still in the queue, waiting at most 2 seconds in total; anything not sent by then is dropped. sdk-garden-python prioritizes zero impact on your program over delivery guarantees. Processes killed by a signal or ended with os._exit() skip the hook. multiprocessing workers started with the fork or forkserver method (the default on Linux) end with os._exit(), so call flush() at the end of their work.

Queue overflow

The internal queue holds a maximum of 100 events. If your SDK generates events faster than the flush thread can drain them, new events are silently discarded. No exception is raised and no warning is logged (unless debug=True).

HTTP timeout

Each connect and read in a request waits at most 2 seconds. A slow endpoint, such as one that trickles its response, can hold the background thread longer, but your code never waits for it: flush(), shutdown() and the send at exit keep their own time limits. After a failed request, the rest of the queue waits for the next flush cycle instead of being sent at once. Each request carries at most batch_size events and at most 512 KiB of them, well under the ingest endpoint's 1 MiB limit.

Usage in test environments

import os
from sdk_garden import SdkGarden

sg = SdkGarden(
    api_key='sgk_test',
    sdk_name='my-sdk',
    sdk_version='0.0.0',
    disabled=os.environ.get('ENV') == 'test',
).start()

When disabled=True:

No thread is spawned.
No network calls are made.
track_method returns the original function unwrapped.
All other methods are no-ops.

Type annotations

The package ships with full type annotations compatible with mypy and pyright:

from sdk_garden import SdkGarden
from typing import Any

def init_observability(api_key: str, sdk_name: str, sdk_version: str) -> SdkGarden:
    return SdkGarden(
        api_key=api_key,
        sdk_name=sdk_name,
        sdk_version=sdk_version,
    ).start()

Installation​

Configuration​

API​

SdkGarden(api_key, sdk_name, sdk_version, **kwargs)​

.start()​

.track(event_name, properties=None)​

.track_method(method_name)​

.flush(timeout=2.0)​

.shutdown(timeout=2.0)​

.track_error(error, context=None)​

Behavior guarantees​

Silent network failure​

Daemon thread exit​

Queue overflow​

HTTP timeout​

Usage in test environments​

Type annotations​

Installation

Configuration

API

`SdkGarden(api_key, sdk_name, sdk_version, **kwargs)`

`.start()`

`.track(event_name, properties=None)`

`.track_method(method_name)`

`.flush(timeout=2.0)`

`.shutdown(timeout=2.0)`

`.track_error(error, context=None)`

Behavior guarantees

Silent network failure

Daemon thread exit

Queue overflow

HTTP timeout

Usage in test environments

Type annotations