Python SDK Deployment Cookbook¶
Practical recipes for real-world extensions. Mix and match the snippets below to accelerate your next integration.
Multi-version handlers¶
Ship breaking changes safely by registering versioned handlers. The SDK enforces that callers specify a version via the header, query param, or URL path.
from kiket_sdk import KiketSDK
sdk = KiketSDK(webhook_secret="secret", workspace_token="wk_test")
@sdk.webhook("campaign.status_changed", version="2024-10-01")
async def v1(payload, context):
await context.endpoints.log_event(
"Campaign status updated",
status=payload["status"],
version=context.event_version,
)
return {"status": payload["status"], "version": "2024-10-01"}
@sdk.webhook("campaign.status_changed", version="2025-02-15")
async def v2(payload, context):
await context.endpoints.emit_metric("campaign.status_changed", 1)
return {
"status": payload["status"],
"changed_by": payload["actor"]["email"],
"version": context.event_version,
}
Secret management¶
Leverage the built-in secret manager to read or rotate credentials without leaving your handler.
@sdk.webhook("campaign.approved", version="v1")
async def handle(payload, context):
api_key = await context.secrets.get("AUTOMATION_API_KEY")
client = httpx.AsyncClient(headers={"Authorization": f"Bearer {api_key.value}"})
await client.post("https://automation.example.com/campaigns", json=payload["campaign"])
await context.endpoints.log_event("automation.triggered", campaign_id=payload["campaign"]["id"])
Creating/rotating secrets:
await context.secrets.set("AUTOMATION_API_KEY", value="sk_live_123")
await context.secrets.rotate("AUTOMATION_API_KEY", value="sk_new_456")
await context.secrets.delete("LEGACY_TOKEN")
Testing with pytest¶
The SDK exposes a FastAPI test client so you can exercise handlers without deploying infrastructure.
import json
import pytest
from kiket_sdk import KiketSDK
@pytest.fixture()
def sdk():
sdk = KiketSDK(webhook_secret="secret", workspace_token="wk_test")
@sdk.webhook("issue.created", version="v1")
async def handler(payload, context):
assert payload["issue"]["id"] == 1
return {"ok": True}
return sdk
def sign(payload, secret):
import hmac, hashlib, datetime, json
body = json.dumps(payload)
signature = hmac.new(secret.encode(), body.encode(), hashlib.sha256).hexdigest()
return {
"body": body,
"headers": {
"Content-Type": "application/json",
"X-Kiket-Signature": signature,
"X-Kiket-Timestamp": datetime.datetime.utcnow().isoformat() + "Z",
"X-Kiket-Event-Version": "v1",
}
}
def test_handler_invocation(sdk):
client = sdk.create_test_client()
signed = sign({"issue": {"id": 1}}, "secret")
response = client.post("/webhooks/issue.created", data=signed["body"], headers=signed["headers"])
assert response.status_code == 200
assert response.json() == {"ok": True}
Telemetry feedback hook¶
Capture handler performance data locally, forward it to your observability stack, or send it back to Kiket via KIKET_SDK_TELEMETRY_URL.
from kiket_sdk import KiketSDK, TelemetryRecord
async def log_record(record: TelemetryRecord) -> None:
print(
f"[sdk] {record.event}@{record.version} "
f"{record.status} in {record.duration_ms:.2f}ms"
)
sdk = KiketSDK(
webhook_secret="secret",
workspace_token="wk_test",
feedback_hook=log_record,
)
Set KIKET_SDK_TELEMETRY_OPTOUT=1 when you need to disable telemetry for air-gapped environments.
Outbound retries¶
context.client is a thin wrapper around httpx.AsyncClient that automatically injects workspace tokens. Use it for all outbound calls to the Kiket API.
async with context.client as client:
await client.post(
"/api/v1/extensions/notifications",
json={
"title": "Campaign approved",
"body": f"{payload['campaign']['name']} is live 🎉",
"level": "info",
},
)
Combine the snippets above to build full-featured extensions with zero boilerplate. For a step-by-step build, continue with the end-to-end tutorial.