Skip to main content

​
Overview

The events client provides real-time streaming of authorization events via Server-Sent Events (SSE). Consume events with an async generator via stream(), or use the subscribe() convenience wrapper for callback-based consumption. Access the events client via client.events.

​
Stream

Open an SSE connection and yield authorization events as an async generator. The connection remains open until you break out of the loop or the server closes it. 
from grantex import Grantex

async with Grantex(api_key="gx_live_...") as client:
    async for event in client.events.stream():
        print(f"Type: {event.type}")
        print(f"ID: {event.id}")
        print(f"Timestamp: {event.timestamp}")
        print(f"Data: {event.data}")

        if event.type == "budget.exhausted":
            break

​
Parameters

ParameterTypeRequiredDescription
event_typeslist[str] | NoneNoFilter to specific event types. If omitted, all types are streamed.
sincestr | NoneNoISO 8601 timestamp to replay events from (for catch-up after disconnection).

​
GrantexEvent

FieldTypeDescription
idstrUnique event identifier.
typestrEvent type (see Event types below).
timestampstrISO 8601 timestamp when the event occurred.
datadictEvent payload. Shape varies by event type.

​
Subscribe

Subscribe to events with a callback function. Returns an unsubscribe callable to close the connection. This is a convenience wrapper around stream(). 
from grantex import Grantex

async with Grantex(api_key="gx_live_...") as client:
    def handle_event(event):
        if event.type == "grant.created":
            print(f"New grant: {event.data['grantId']}")
        elif event.type == "grant.revoked":
            print(f"Grant revoked: {event.data['grantId']}")
        elif event.type == "budget.threshold":
            print(f"Budget at {event.data['percentage']}%")
        elif event.type == "budget.exhausted":
            print(f"Budget exhausted for {event.data['grantId']}")

    unsubscribe = client.events.subscribe(handle_event)

    # Later: close the SSE connection
    unsubscribe()

​
Parameters

ParameterTypeRequiredDescription
handlerCallable[[GrantexEvent], None]YesCallback invoked for each event.
event_typeslist[str] | NoneNoFilter to specific event types.
sincestr | NoneNoISO 8601 timestamp to replay events from.

​
Returns

Returns a callable. Call it to close the SSE connection and stop receiving events.

​
Event types

​
grant.created

Emitted when a new grant is created after user consent. 
{
    "type": "grant.created",
    "data": {
        "grantId": "grnt_01HXYZ...",
        "agentId": "ag_01HXYZ...",
        "principalId": "user_abc123",
        "scopes": ["calendar:read", "contacts:read"],
    }
}

​
grant.revoked

Emitted when a grant is revoked. 
{
    "type": "grant.revoked",
    "data": {
        "grantId": "grnt_01HXYZ...",
        "revokedBy": "developer",  # 'developer' | 'principal' | 'system'
    }
}

​
token.issued

Emitted when a grant token is issued (via exchange or refresh). 
{
    "type": "token.issued",
    "data": {
        "grantId": "grnt_01HXYZ...",
        "tokenId": "tok_01HXYZ...",
        "method": "exchange",  # 'exchange' | 'refresh'
    }
}

​
budget.threshold

Emitted when a grant’s budget consumption crosses 50% or 80%. 
{
    "type": "budget.threshold",
    "data": {
        "grantId": "grnt_01HXYZ...",
        "percentage": 80,
        "remaining": 200,
        "initial": 1000,
    }
}

​
budget.exhausted

Emitted when a grant’s budget is fully consumed. 
{
    "type": "budget.exhausted",
    "data": {
        "grantId": "grnt_01HXYZ...",
        "initial": 1000,
    }
}

​
Example: Budget Alert Monitor

import asyncio
from grantex import Grantex

async def monitor_budgets():
    async with Grantex(api_key="gx_live_...") as client:
        async for event in client.events.stream(
            event_types=["budget.threshold", "budget.exhausted"]
        ):
            if event.type == "budget.threshold":
                print(
                    f"Budget alert: {event.data['percentage']}% used "
                    f"for grant {event.data['grantId']}"
                )
            elif event.type == "budget.exhausted":
                print(f"Budget exhausted for grant {event.data['grantId']}")
                # Revoke the grant to prevent further usage
                client.grants.revoke(event.data["grantId"])

asyncio.run(monitor_budgets())