streamgine — Python SDK

Real-time webhook events for California corporate registry changes. Get notified the moment a company is added, updated, or removed — as structured JSON diffs, delivered to your endpoint.

Product: diff.streamgine.com · Docs: docs.md · Coverage: California · Use cases: risk monitoring, AI agents

Install

pip install streamgine

Requires Python 3.10+. One dependency: httpx.

Quickstart

Get your credentials from diff.streamgine.com/account, then:

from streamgine import diff

diff.configure("diff-streamgine-credentials.json")
# or: source diff-streamgine-env.sh   then   diff.configure()

client = diff.DiffClient()

1. Confirm your webhook is reachable

Before wiring real events, send yourself a test heartbeat:

client.register_test(callback_url="https://your-agent.example/webhooks/diff")

You'll start receiving a signed test event every second on topic diff-event.test:

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

Filter on event.is_test_heartbeat to ignore these once you're live.

2. Subscribe to real events

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

INSERT = new company, UPDATE = changed, DELETE = removed. Omit actions to receive all three.

3. Search current company records

Webhooks deliver changes over time. search() returns current snapshots:

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

4. Verify inbound webhooks (required)

Always verify raw request bytes before trusting the payload:

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():
    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"}
    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 your signing secret, and never skip verification in production.

Credentials

Download from diff.streamgine.com/account:

Call configure() once at startup — every DiffClient() and verify_webhook() call reads from it.

API surface (streamgine.diff)

Links

• Documentation
• Homepage
• Issues
• Changelog