streamgine — Python SDK

Diff by Streamgine lives under the diff module. Install once; add more Streamgine modules (training data, etc.) under the same package later.

This directory is the customer package published to PyPI. Repo-only load tests and dev agents live in tools/python/ — not shipped to customers.

Today: one wheel streamgine = Diff (from streamgine import diff). Next products (e.g. training-data) get their own wheels (streamgine-diff, streamgine-training, …) under the same streamgine import namespace — see PUBLISHING.md and .cursor/rules/python-client.mdc. Do not grow a fat single package.

Product: diff.streamgine.com · Docs: docs.md · California: coverage · Use cases: risk monitoring, AI agents · Repo: github.com/scurtutech/diff-events-service

from streamgine import diff

# Once at startup — pick ONE download from /account:
diff.configure("diff-streamgine-credentials.json")
# OR: source diff-streamgine-env.sh  then  diff.configure()
client = diff.DiffClient()

Install from PyPI:

python3 -m pip install streamgine

Or from this repo:

cd clients/python
python3 -m venv .venv && source .venv/bin/activate
python -m pip install .

Requires Python 3.10+. One dependency: httpx (TLS verification enabled by default).

Credentials (one setup point)

Call diff.configure() once at process startup. Every DiffClient() and verify_webhook() reads from that config.

Download from /account or checkout — filenames are fixed:

Option A — credentials JSON:

diff.configure("diff-streamgine-credentials.json")

Option B — env script:

source diff-streamgine-env.sh

diff.configure()

Do not pass API keys or signing secrets to DiffClient() or verify_webhook() in normal use.

Zero-setup agent (copy-paste)

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -e . fastapi uvicorn
# Customer path: download from /account, then ONE of:
#   source diff-streamgine-env.sh
#   # or keep diff-streamgine-credentials.json next to the process
# Local repo seed (AUTH_ENABLED): npm run seed:dev-customer, then export printed keys
source diff-streamgine-env.sh   # or configure("diff-streamgine-credentials.json") in code
python -m uvicorn examples.agent:app --port 8000

Runnable source: examples/agent.py. Calls register_test() on startup; within ~1s you should see after={'test': 'success'} on topic diff-event.test.

Serverless (AWS Lambda): examples/serverless/ — deploy handler.py, run register.py once, receive test heartbeats in CloudWatch.

Agent journey (production filters): examples/agent_journey.py.

Docker worker reaching a host agent:

export DIFF_CALLBACK_URL=http://host.docker.internal:8000/webhooks/diff

Test heartbeats (start here)

Before wiring real registry events, confirm your agent receives signed webhooks:

from streamgine import diff

diff.configure()  # after: source diff-streamgine-env.sh  OR  configure("diff-streamgine-credentials.json")
client = diff.DiffClient()
client.register_test(callback_url="https://your-agent.example/webhooks/diff")

Filter on event.is_test_heartbeat (or after == {"test": "success"}). The service sends this every second:

{
  "action": "INSERT",
  "after": { "test": "success" }
}

No entity_id on test events.

Quick start (production registry events)

from streamgine import diff

diff.configure()
client = diff.DiffClient()
client.register(
    match={
        "state": "CA",
        "principal_city": "San Francisco",
    },
    actions=["INSERT"],
    callback_url="https://your-agent.example/webhooks/diff",
)
client.close()

INSERT = newly added company, UPDATE = changed, DELETE = removed. Omit actions to receive every matching state/city event.

Search current companies (full story)

Webhooks deliver changes. Search returns current snapshots:

from streamgine import diff

diff.configure()
client = diff.DiffClient()
result = client.search(
    query={"match": {"principalCity": "San Francisco"}},
    size=10,
)
for company in result["entities"]:
    print(company.get("entity_name"), company.get("entity_number"))

Agent journey: Search → Watch → Qualify → Act → Stay current (see /docs.md).

2. Verify inbound webhooks (required)

Always verify raw request bytes before trusting JSON. The worker signs the exact JSON body with HMAC-SHA256.

from streamgine import diff

diff.configure()

def handle_webhook(raw_body: bytes, signature: str | None):
    try:
        event = diff.verify_webhook(raw_body, signature)
    except diff.WebhookVerificationError:
        return 401, {"error": "invalid signature"}

    print(event.action, event.entity_id, event.after)
    return 200, {"ok": True}

FastAPI

from fastapi import FastAPI, Header, Request, Response

from streamgine import diff

app = FastAPI()

@app.on_event("startup")
def setup() -> None:
    diff.configure()

@app.post("/webhooks/diff")
async def diff_webhook(
    request: Request,
    response: Response,
    x_diff_signature: str | None = Header(default=None, alias="X-Diff-Signature"),
):
    raw = await request.body()
    try:
        event = diff.verify_webhook(raw, x_diff_signature)
    except diff.WebhookVerificationError:
        response.status_code = 401
        return {"error": "invalid signature"}

    # Your agent logic here
    return {"ok": True, "entity_id": event.entity_id}

Flask

from flask import Flask, request

from streamgine import diff

app = Flask(__name__)
diff.configure()

@app.post("/webhooks/diff")
def webhook():
    try:
        event = diff.verify_webhook(
            request.get_data(),
            request.headers.get("X-Diff-Signature"),
        )
    except diff.WebhookVerificationError:
        return {"error": "invalid signature"}, 401

    return {"ok": True, "action": event.action}

Security

Never log the signing secret or skip signature verification in production.

Credentials file (after checkout)

Download diff-streamgine-credentials.json (or diff-streamgine-env.sh) from /account / checkout. JSON format (version 1):

{
  "version": 1,
  "service": "diff.streamgine.com",
  "customer_id": "cus_…",
  "api_key": "de_live_…",
  "signing_secret": "whsec_…",
  "api_url": "https://diff.streamgine.com",
  "env": {
    "DIFF_API_KEY": "de_live_…",
    "DIFF_SIGNING_SECRET": "whsec_…",
    "DIFF_API_URL": "https://diff.streamgine.com"
  }
}

from streamgine import diff

diff.configure("diff-streamgine-credentials.json")
client = diff.DiffClient()
# diff.verify_webhook(...) uses the same signing secret automatically

Auth

Register responses are slim (ok / subscription fields only — no Redis index maps).

API surface (streamgine.diff)

Direct imports also work: from streamgine.diff import DiffClient.

Development

cd clients/python
python -m pip install -e ".[dev]"
pytest