webex_byova.auth
webex_byova.auth
Authentication and token management.
DEFAULT_INTEGRATION_SCOPES
module-attribute
DEFAULT_INTEGRATION_SCOPES: list[str] = ['spark:all', 'spark:applications_token', 'application:webhooks_write', 'application:webhooks_read']
Default OAuth scopes required for BYOVA Integration authorization.
__all__
module-attribute
__all__ = ['DEFAULT_INTEGRATION_SCOPES', 'IntegrationTokenManager', 'ServiceAppTokenManager', 'TokenStorage', 'InMemoryTokenStorage', 'derive_application_id', 'decode_org_id', 'load_credentials_from_env']
IntegrationTokenManager
Manage Integration OAuth tokens for the developer-initiated flow.
Handles authorization URL generation, code exchange, token refresh, and the full browser-based OAuth flow with a local redirect listener.
credentials
property
credentials: IntegrationCredentials
Integration OAuth client credentials.
__init__
__init__(credentials: IntegrationCredentials, http: HttpClient, storage: TokenStorage, config: BYOVAConfig | None = None) -> None
Initialize the Integration token manager.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
credentials
|
IntegrationCredentials
|
Integration OAuth client credentials. |
required |
http
|
HttpClient
|
Shared HTTP client for API requests. |
required |
storage
|
TokenStorage
|
Token storage backend. |
required |
config
|
BYOVAConfig | None
|
Optional SDK configuration; defaults to |
None
|
get_authorization_url
get_authorization_url(scopes: list[str] | None = None, *, state: str | None = None) -> tuple[str, str]
Build the OAuth authorization URL.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
scopes
|
list[str] | None
|
OAuth scopes to request; defaults to |
None
|
state
|
str | None
|
CSRF state token; generated automatically if not provided. |
None
|
Returns:
| Type | Description |
|---|---|
tuple[str, str]
|
Tuple of |
exchange_code
exchange_code(code: str) -> OAuthTokens
Exchange an authorization code for OAuth tokens (sync).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
code
|
str
|
Authorization code from the OAuth redirect. |
required |
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
OAuth tokens from the Webex token endpoint. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If the token exchange fails. |
aexchange_code
async
aexchange_code(code: str) -> OAuthTokens
Exchange an authorization code for OAuth tokens (async).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
code
|
str
|
Authorization code from the OAuth redirect. |
required |
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
OAuth tokens from the Webex token endpoint. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If the token exchange fails. |
refresh
refresh(refresh_token: str | None = None) -> OAuthTokens
Refresh the Integration access token (sync wrapper).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
refresh_token
|
str | None
|
Explicit refresh token; uses stored token if omitted. |
None
|
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
Refreshed OAuth tokens. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If no refresh token is available. |
arefresh
async
arefresh(refresh_token: str | None = None) -> OAuthTokens
Refresh the Integration access token.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
refresh_token
|
str | None
|
Explicit refresh token; uses stored token if omitted. |
None
|
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
Refreshed OAuth tokens, persisted to storage. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If no refresh token is available. |
authorize
authorize(scopes: list[str] | None = None, *, open_browser: bool = True, timeout: float = 300.0, state: str | None = None) -> OAuthTokens
Run the full OAuth flow with a local redirect listener (sync).
Opens the authorization URL in a browser, waits for the redirect, exchanges the code, and stores tokens.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
scopes
|
list[str] | None
|
OAuth scopes to request. |
None
|
open_browser
|
bool
|
Whether to open the authorization URL automatically. |
True
|
timeout
|
float
|
Maximum seconds to wait for the OAuth redirect. |
300.0
|
state
|
str | None
|
CSRF state token; generated if not provided. |
None
|
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
OAuth tokens from the completed authorization. |
Raises:
| Type | Description |
|---|---|
OAuthRedirectError
|
If the user denies access. |
OAuthRedirectTimeout
|
If the redirect listener times out. |
AuthenticationError
|
If token exchange fails. |
aauthorize
async
aauthorize(scopes: list[str] | None = None, *, open_browser: bool = True, timeout: float = 300.0, state: str | None = None) -> OAuthTokens
Run the full OAuth flow with a local redirect listener (async).
Opens the authorization URL in a browser, waits for the redirect, exchanges the code, and stores tokens.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
scopes
|
list[str] | None
|
OAuth scopes to request. |
None
|
open_browser
|
bool
|
Whether to open the authorization URL automatically. |
True
|
timeout
|
float
|
Maximum seconds to wait for the OAuth redirect. |
300.0
|
state
|
str | None
|
CSRF state token; generated if not provided. |
None
|
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
OAuth tokens from the completed authorization. |
Raises:
| Type | Description |
|---|---|
OAuthRedirectError
|
If the user denies access. |
OAuthRedirectTimeout
|
If the redirect listener times out. |
AuthenticationError
|
If token exchange fails. |
get_access_token
get_access_token() -> str
Return a valid Integration access token, refreshing if needed (sync).
Returns:
| Type | Description |
|---|---|
str
|
Valid bearer access token. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If Integration is not yet authorized. |
aget_access_token
async
aget_access_token() -> str
Return a valid Integration access token, refreshing if needed.
Returns:
| Type | Description |
|---|---|
str
|
Valid bearer access token. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If Integration is not yet authorized. |
ServiceAppTokenManager
Fetch, refresh, and store per-organization Service App tokens.
Service App tokens are scoped to a customer organization and are obtained after a customer admin authorizes your Service App in Control Hub.
application_id
property
application_id: str
Base64url-encoded Service App application ID derived from client ID.
credentials
property
credentials: ServiceAppCredentials
Service App OAuth client credentials.
__init__
__init__(credentials: ServiceAppCredentials, integration: IntegrationTokenManager, http: HttpClient, storage: TokenStorage) -> None
Initialize the Service App token manager.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
credentials
|
ServiceAppCredentials
|
Service App OAuth client credentials. |
required |
integration
|
IntegrationTokenManager
|
Integration token manager for bearer authentication. |
required |
http
|
HttpClient
|
Shared HTTP client for API requests. |
required |
storage
|
TokenStorage
|
Token storage backend. |
required |
afetch_token_for_org
async
afetch_token_for_org(org_id: str) -> ServiceAppTokens
Fetch Service App tokens for an organization using Integration bearer auth.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Target organization UUID. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens
|
Service App tokens, persisted to storage. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If the token fetch fails. |
fetch_token_for_org
fetch_token_for_org(org_id: str) -> ServiceAppTokens
Fetch Service App tokens for an organization (sync wrapper).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Target organization UUID. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens
|
Service App tokens, persisted to storage. |
afetch_token_from_webhook_org
async
afetch_token_from_webhook_org(encoded_org_id: str) -> tuple[str, ServiceAppTokens]
Decode org ID from a webhook payload and fetch Service App tokens.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
encoded_org_id
|
str
|
Base64url-encoded org ID from the webhook |
required |
Returns:
| Type | Description |
|---|---|
tuple[str, ServiceAppTokens]
|
Tuple of |
save_registration
save_registration(org_id: str, refresh_token: str) -> ServiceAppTokens
Store tokens when a refresh token is already available (sync).
Use this sandbox path when you already have a Service App refresh token without waiting for a production webhook.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
refresh_token
|
str
|
Existing Service App refresh token. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens
|
Refreshed Service App tokens. |
asave_registration
async
asave_registration(org_id: str, refresh_token: str) -> ServiceAppTokens
Store tokens when a refresh token is already available.
Use this sandbox path when you already have a Service App refresh token without waiting for a production webhook.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
refresh_token
|
str
|
Existing Service App refresh token. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens
|
Refreshed Service App tokens. |
arefresh_for_org
async
arefresh_for_org(org_id: str) -> ServiceAppTokens
Refresh the Service App access token for an organization.
If no refresh token is stored, falls back to fetching new tokens via the Integration bearer.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens
|
Refreshed Service App tokens, persisted to storage. |
aget_access_token
async
aget_access_token(org_id: str) -> str
Return a valid Service App access token for an organization.
Refreshes automatically if the token is missing or expired.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
Returns:
| Type | Description |
|---|---|
str
|
Valid bearer access token for org-scoped API calls. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If the org is not registered. |
InMemoryTokenStorage
Default in-process token storage for development and testing.
Tokens are lost when the process exits. Use a custom TokenStorage
implementation for production deployments.
__init__
__init__() -> None
Initialize empty in-memory token stores.
get_integration_tokens
async
get_integration_tokens() -> OAuthTokens | None
Return stored Integration OAuth tokens.
Returns:
| Type | Description |
|---|---|
OAuthTokens | None
|
Integration tokens, or |
set_integration_tokens
async
set_integration_tokens(tokens: OAuthTokens) -> None
Store Integration OAuth tokens.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tokens
|
OAuthTokens
|
Tokens to persist. |
required |
get_service_app_tokens
async
get_service_app_tokens(org_id: str) -> ServiceAppTokens | None
Return Service App tokens for an organization.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens | None
|
Service App tokens, or |
set_service_app_tokens
async
set_service_app_tokens(org_id: str, tokens: ServiceAppTokens) -> None
Store Service App tokens for an organization.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
tokens
|
ServiceAppTokens
|
Tokens to persist. |
required |
delete_service_app_tokens
async
delete_service_app_tokens(org_id: str) -> None
Remove Service App tokens for an organization.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
list_registered_orgs
async
list_registered_orgs() -> list[str]
Return all organization IDs with stored Service App tokens.
Returns:
| Type | Description |
|---|---|
list[str]
|
List of registered organization UUIDs. |
TokenStorage
Bases: Protocol
Protocol for storing Integration and per-org Service App tokens.
Implement this protocol for production persistence (Redis, database, etc.). All methods are async to support non-blocking I/O backends.
get_integration_tokens
async
get_integration_tokens() -> OAuthTokens | None
Return stored Integration OAuth tokens, or None if not authorized.
set_integration_tokens
async
set_integration_tokens(tokens: OAuthTokens) -> None
Persist Integration OAuth tokens.
get_service_app_tokens
async
get_service_app_tokens(org_id: str) -> ServiceAppTokens | None
Return Service App tokens for an org, or None if not registered.
set_service_app_tokens
async
set_service_app_tokens(org_id: str, tokens: ServiceAppTokens) -> None
Persist Service App tokens for an organization.
delete_service_app_tokens
async
delete_service_app_tokens(org_id: str) -> None
Remove Service App tokens for an organization (deauthorization).
list_registered_orgs
async
list_registered_orgs() -> list[str]
Return organization IDs with stored Service App tokens.
load_credentials_from_env
load_credentials_from_env() -> tuple[IntegrationCredentials, ServiceAppCredentials]
Load Integration and Service App credentials from environment variables.
Required variables
WEBEX_INTEGRATION_CLIENT_IDWEBEX_INTEGRATION_CLIENT_SECRETWEBEX_SA_CLIENT_IDWEBEX_SA_CLIENT_SECRET
Optional variables
WEBEX_INTEGRATION_REDIRECT_URI(defaults tohttp://127.0.0.1:8765/callback)
Returns:
| Type | Description |
|---|---|
tuple[IntegrationCredentials, ServiceAppCredentials]
|
Tuple of |
Raises:
| Type | Description |
|---|---|
ValueError
|
If any required environment variable is missing. |
decode_org_id
decode_org_id(encoded_org_id: str) -> str
Decode an organization ID from a webhook orgId value.
Webex sends a base64url-encoded URI in authorized webhook payloads. This extracts the org UUID from the decoded path.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
encoded_org_id
|
str
|
Base64url-encoded org identifier from a webhook. |
required |
Returns:
| Type | Description |
|---|---|
str
|
Decoded organization UUID, or the full decoded string if parsing fails. |
derive_application_id
derive_application_id(service_app_client_id: str) -> str
Derive the Service App application ID from its client ID.
Webex encodes the application ID as base64url of
ciscospark://us/APPLICATION/{client_id}.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
service_app_client_id
|
str
|
Service App OAuth client ID. |
required |
Returns:
| Type | Description |
|---|---|
str
|
Base64url-encoded application ID used in API paths. |
credentials
Credential loading helpers.
load_credentials_from_env
load_credentials_from_env() -> tuple[IntegrationCredentials, ServiceAppCredentials]
Load Integration and Service App credentials from environment variables.
Required variables
WEBEX_INTEGRATION_CLIENT_IDWEBEX_INTEGRATION_CLIENT_SECRETWEBEX_SA_CLIENT_IDWEBEX_SA_CLIENT_SECRET
Optional variables
WEBEX_INTEGRATION_REDIRECT_URI(defaults tohttp://127.0.0.1:8765/callback)
Returns:
| Type | Description |
|---|---|
tuple[IntegrationCredentials, ServiceAppCredentials]
|
Tuple of |
Raises:
| Type | Description |
|---|---|
ValueError
|
If any required environment variable is missing. |
integration
Integration OAuth token management.
DEFAULT_INTEGRATION_SCOPES
module-attribute
DEFAULT_INTEGRATION_SCOPES: list[str] = ['spark:all', 'spark:applications_token', 'application:webhooks_write', 'application:webhooks_read']
Default OAuth scopes required for BYOVA Integration authorization.
IntegrationTokenManager
Manage Integration OAuth tokens for the developer-initiated flow.
Handles authorization URL generation, code exchange, token refresh, and the full browser-based OAuth flow with a local redirect listener.
credentials
property
credentials: IntegrationCredentials
Integration OAuth client credentials.
__init__
__init__(credentials: IntegrationCredentials, http: HttpClient, storage: TokenStorage, config: BYOVAConfig | None = None) -> None
Initialize the Integration token manager.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
credentials
|
IntegrationCredentials
|
Integration OAuth client credentials. |
required |
http
|
HttpClient
|
Shared HTTP client for API requests. |
required |
storage
|
TokenStorage
|
Token storage backend. |
required |
config
|
BYOVAConfig | None
|
Optional SDK configuration; defaults to |
None
|
get_authorization_url
get_authorization_url(scopes: list[str] | None = None, *, state: str | None = None) -> tuple[str, str]
Build the OAuth authorization URL.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
scopes
|
list[str] | None
|
OAuth scopes to request; defaults to |
None
|
state
|
str | None
|
CSRF state token; generated automatically if not provided. |
None
|
Returns:
| Type | Description |
|---|---|
tuple[str, str]
|
Tuple of |
exchange_code
exchange_code(code: str) -> OAuthTokens
Exchange an authorization code for OAuth tokens (sync).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
code
|
str
|
Authorization code from the OAuth redirect. |
required |
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
OAuth tokens from the Webex token endpoint. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If the token exchange fails. |
aexchange_code
async
aexchange_code(code: str) -> OAuthTokens
Exchange an authorization code for OAuth tokens (async).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
code
|
str
|
Authorization code from the OAuth redirect. |
required |
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
OAuth tokens from the Webex token endpoint. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If the token exchange fails. |
refresh
refresh(refresh_token: str | None = None) -> OAuthTokens
Refresh the Integration access token (sync wrapper).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
refresh_token
|
str | None
|
Explicit refresh token; uses stored token if omitted. |
None
|
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
Refreshed OAuth tokens. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If no refresh token is available. |
arefresh
async
arefresh(refresh_token: str | None = None) -> OAuthTokens
Refresh the Integration access token.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
refresh_token
|
str | None
|
Explicit refresh token; uses stored token if omitted. |
None
|
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
Refreshed OAuth tokens, persisted to storage. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If no refresh token is available. |
authorize
authorize(scopes: list[str] | None = None, *, open_browser: bool = True, timeout: float = 300.0, state: str | None = None) -> OAuthTokens
Run the full OAuth flow with a local redirect listener (sync).
Opens the authorization URL in a browser, waits for the redirect, exchanges the code, and stores tokens.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
scopes
|
list[str] | None
|
OAuth scopes to request. |
None
|
open_browser
|
bool
|
Whether to open the authorization URL automatically. |
True
|
timeout
|
float
|
Maximum seconds to wait for the OAuth redirect. |
300.0
|
state
|
str | None
|
CSRF state token; generated if not provided. |
None
|
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
OAuth tokens from the completed authorization. |
Raises:
| Type | Description |
|---|---|
OAuthRedirectError
|
If the user denies access. |
OAuthRedirectTimeout
|
If the redirect listener times out. |
AuthenticationError
|
If token exchange fails. |
aauthorize
async
aauthorize(scopes: list[str] | None = None, *, open_browser: bool = True, timeout: float = 300.0, state: str | None = None) -> OAuthTokens
Run the full OAuth flow with a local redirect listener (async).
Opens the authorization URL in a browser, waits for the redirect, exchanges the code, and stores tokens.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
scopes
|
list[str] | None
|
OAuth scopes to request. |
None
|
open_browser
|
bool
|
Whether to open the authorization URL automatically. |
True
|
timeout
|
float
|
Maximum seconds to wait for the OAuth redirect. |
300.0
|
state
|
str | None
|
CSRF state token; generated if not provided. |
None
|
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
OAuth tokens from the completed authorization. |
Raises:
| Type | Description |
|---|---|
OAuthRedirectError
|
If the user denies access. |
OAuthRedirectTimeout
|
If the redirect listener times out. |
AuthenticationError
|
If token exchange fails. |
get_access_token
get_access_token() -> str
Return a valid Integration access token, refreshing if needed (sync).
Returns:
| Type | Description |
|---|---|
str
|
Valid bearer access token. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If Integration is not yet authorized. |
aget_access_token
async
aget_access_token() -> str
Return a valid Integration access token, refreshing if needed.
Returns:
| Type | Description |
|---|---|
str
|
Valid bearer access token. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If Integration is not yet authorized. |
redirect_listener
Local HTTP server for OAuth authorization code redirect.
wait_for_redirect
wait_for_redirect(redirect_uri: str, *, timeout: float = 300.0, open_browser: bool = False, authorization_url: str | None = None, expected_state: str | None = None) -> tuple[str, str | None]
Start a temporary HTTP server and wait for OAuth redirect with ?code=.
Returns (authorization_code, state).
service_app
Service App token management.
ServiceAppTokenManager
Fetch, refresh, and store per-organization Service App tokens.
Service App tokens are scoped to a customer organization and are obtained after a customer admin authorizes your Service App in Control Hub.
application_id
property
application_id: str
Base64url-encoded Service App application ID derived from client ID.
credentials
property
credentials: ServiceAppCredentials
Service App OAuth client credentials.
__init__
__init__(credentials: ServiceAppCredentials, integration: IntegrationTokenManager, http: HttpClient, storage: TokenStorage) -> None
Initialize the Service App token manager.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
credentials
|
ServiceAppCredentials
|
Service App OAuth client credentials. |
required |
integration
|
IntegrationTokenManager
|
Integration token manager for bearer authentication. |
required |
http
|
HttpClient
|
Shared HTTP client for API requests. |
required |
storage
|
TokenStorage
|
Token storage backend. |
required |
afetch_token_for_org
async
afetch_token_for_org(org_id: str) -> ServiceAppTokens
Fetch Service App tokens for an organization using Integration bearer auth.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Target organization UUID. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens
|
Service App tokens, persisted to storage. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If the token fetch fails. |
fetch_token_for_org
fetch_token_for_org(org_id: str) -> ServiceAppTokens
Fetch Service App tokens for an organization (sync wrapper).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Target organization UUID. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens
|
Service App tokens, persisted to storage. |
afetch_token_from_webhook_org
async
afetch_token_from_webhook_org(encoded_org_id: str) -> tuple[str, ServiceAppTokens]
Decode org ID from a webhook payload and fetch Service App tokens.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
encoded_org_id
|
str
|
Base64url-encoded org ID from the webhook |
required |
Returns:
| Type | Description |
|---|---|
tuple[str, ServiceAppTokens]
|
Tuple of |
save_registration
save_registration(org_id: str, refresh_token: str) -> ServiceAppTokens
Store tokens when a refresh token is already available (sync).
Use this sandbox path when you already have a Service App refresh token without waiting for a production webhook.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
refresh_token
|
str
|
Existing Service App refresh token. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens
|
Refreshed Service App tokens. |
asave_registration
async
asave_registration(org_id: str, refresh_token: str) -> ServiceAppTokens
Store tokens when a refresh token is already available.
Use this sandbox path when you already have a Service App refresh token without waiting for a production webhook.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
refresh_token
|
str
|
Existing Service App refresh token. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens
|
Refreshed Service App tokens. |
arefresh_for_org
async
arefresh_for_org(org_id: str) -> ServiceAppTokens
Refresh the Service App access token for an organization.
If no refresh token is stored, falls back to fetching new tokens via the Integration bearer.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens
|
Refreshed Service App tokens, persisted to storage. |
aget_access_token
async
aget_access_token(org_id: str) -> str
Return a valid Service App access token for an organization.
Refreshes automatically if the token is missing or expired.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
Returns:
| Type | Description |
|---|---|
str
|
Valid bearer access token for org-scoped API calls. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If the org is not registered. |
storage
Pluggable token storage.
TokenStorage
Bases: Protocol
Protocol for storing Integration and per-org Service App tokens.
Implement this protocol for production persistence (Redis, database, etc.). All methods are async to support non-blocking I/O backends.
get_integration_tokens
async
get_integration_tokens() -> OAuthTokens | None
Return stored Integration OAuth tokens, or None if not authorized.
set_integration_tokens
async
set_integration_tokens(tokens: OAuthTokens) -> None
Persist Integration OAuth tokens.
get_service_app_tokens
async
get_service_app_tokens(org_id: str) -> ServiceAppTokens | None
Return Service App tokens for an org, or None if not registered.
set_service_app_tokens
async
set_service_app_tokens(org_id: str, tokens: ServiceAppTokens) -> None
Persist Service App tokens for an organization.
delete_service_app_tokens
async
delete_service_app_tokens(org_id: str) -> None
Remove Service App tokens for an organization (deauthorization).
list_registered_orgs
async
list_registered_orgs() -> list[str]
Return organization IDs with stored Service App tokens.
InMemoryTokenStorage
Default in-process token storage for development and testing.
Tokens are lost when the process exits. Use a custom TokenStorage
implementation for production deployments.
__init__
__init__() -> None
Initialize empty in-memory token stores.
get_integration_tokens
async
get_integration_tokens() -> OAuthTokens | None
Return stored Integration OAuth tokens.
Returns:
| Type | Description |
|---|---|
OAuthTokens | None
|
Integration tokens, or |
set_integration_tokens
async
set_integration_tokens(tokens: OAuthTokens) -> None
Store Integration OAuth tokens.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tokens
|
OAuthTokens
|
Tokens to persist. |
required |
get_service_app_tokens
async
get_service_app_tokens(org_id: str) -> ServiceAppTokens | None
Return Service App tokens for an organization.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens | None
|
Service App tokens, or |
set_service_app_tokens
async
set_service_app_tokens(org_id: str, tokens: ServiceAppTokens) -> None
Store Service App tokens for an organization.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
tokens
|
ServiceAppTokens
|
Tokens to persist. |
required |
delete_service_app_tokens
async
delete_service_app_tokens(org_id: str) -> None
Remove Service App tokens for an organization.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
list_registered_orgs
async
list_registered_orgs() -> list[str]
Return all organization IDs with stored Service App tokens.
Returns:
| Type | Description |
|---|---|
list[str]
|
List of registered organization UUIDs. |
utils
Webex auth utility functions.
derive_application_id
derive_application_id(service_app_client_id: str) -> str
Derive the Service App application ID from its client ID.
Webex encodes the application ID as base64url of
ciscospark://us/APPLICATION/{client_id}.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
service_app_client_id
|
str
|
Service App OAuth client ID. |
required |
Returns:
| Type | Description |
|---|---|
str
|
Base64url-encoded application ID used in API paths. |
decode_org_id
decode_org_id(encoded_org_id: str) -> str
Decode an organization ID from a webhook orgId value.
Webex sends a base64url-encoded URI in authorized webhook payloads. This extracts the org UUID from the decoded path.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
encoded_org_id
|
str
|
Base64url-encoded org identifier from a webhook. |
required |
Returns:
| Type | Description |
|---|---|
str
|
Decoded organization UUID, or the full decoded string if parsing fails. |
webex_byova.auth.integration.IntegrationTokenManager
Manage Integration OAuth tokens for the developer-initiated flow.
Handles authorization URL generation, code exchange, token refresh, and the full browser-based OAuth flow with a local redirect listener.
_credentials
instance-attribute
_credentials = credentials
_http
instance-attribute
_http = http
_storage
instance-attribute
_storage = storage
_config
instance-attribute
_config = config or config
credentials
property
credentials: IntegrationCredentials
Integration OAuth client credentials.
__init__
__init__(credentials: IntegrationCredentials, http: HttpClient, storage: TokenStorage, config: BYOVAConfig | None = None) -> None
Initialize the Integration token manager.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
credentials
|
IntegrationCredentials
|
Integration OAuth client credentials. |
required |
http
|
HttpClient
|
Shared HTTP client for API requests. |
required |
storage
|
TokenStorage
|
Token storage backend. |
required |
config
|
BYOVAConfig | None
|
Optional SDK configuration; defaults to |
None
|
get_authorization_url
get_authorization_url(scopes: list[str] | None = None, *, state: str | None = None) -> tuple[str, str]
Build the OAuth authorization URL.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
scopes
|
list[str] | None
|
OAuth scopes to request; defaults to |
None
|
state
|
str | None
|
CSRF state token; generated automatically if not provided. |
None
|
Returns:
| Type | Description |
|---|---|
tuple[str, str]
|
Tuple of |
_parse_token_response
_parse_token_response(data: dict) -> OAuthTokens
exchange_code
exchange_code(code: str) -> OAuthTokens
Exchange an authorization code for OAuth tokens (sync).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
code
|
str
|
Authorization code from the OAuth redirect. |
required |
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
OAuth tokens from the Webex token endpoint. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If the token exchange fails. |
aexchange_code
async
aexchange_code(code: str) -> OAuthTokens
Exchange an authorization code for OAuth tokens (async).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
code
|
str
|
Authorization code from the OAuth redirect. |
required |
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
OAuth tokens from the Webex token endpoint. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If the token exchange fails. |
refresh
refresh(refresh_token: str | None = None) -> OAuthTokens
Refresh the Integration access token (sync wrapper).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
refresh_token
|
str | None
|
Explicit refresh token; uses stored token if omitted. |
None
|
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
Refreshed OAuth tokens. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If no refresh token is available. |
arefresh
async
arefresh(refresh_token: str | None = None) -> OAuthTokens
Refresh the Integration access token.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
refresh_token
|
str | None
|
Explicit refresh token; uses stored token if omitted. |
None
|
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
Refreshed OAuth tokens, persisted to storage. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If no refresh token is available. |
authorize
authorize(scopes: list[str] | None = None, *, open_browser: bool = True, timeout: float = 300.0, state: str | None = None) -> OAuthTokens
Run the full OAuth flow with a local redirect listener (sync).
Opens the authorization URL in a browser, waits for the redirect, exchanges the code, and stores tokens.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
scopes
|
list[str] | None
|
OAuth scopes to request. |
None
|
open_browser
|
bool
|
Whether to open the authorization URL automatically. |
True
|
timeout
|
float
|
Maximum seconds to wait for the OAuth redirect. |
300.0
|
state
|
str | None
|
CSRF state token; generated if not provided. |
None
|
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
OAuth tokens from the completed authorization. |
Raises:
| Type | Description |
|---|---|
OAuthRedirectError
|
If the user denies access. |
OAuthRedirectTimeout
|
If the redirect listener times out. |
AuthenticationError
|
If token exchange fails. |
aauthorize
async
aauthorize(scopes: list[str] | None = None, *, open_browser: bool = True, timeout: float = 300.0, state: str | None = None) -> OAuthTokens
Run the full OAuth flow with a local redirect listener (async).
Opens the authorization URL in a browser, waits for the redirect, exchanges the code, and stores tokens.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
scopes
|
list[str] | None
|
OAuth scopes to request. |
None
|
open_browser
|
bool
|
Whether to open the authorization URL automatically. |
True
|
timeout
|
float
|
Maximum seconds to wait for the OAuth redirect. |
300.0
|
state
|
str | None
|
CSRF state token; generated if not provided. |
None
|
Returns:
| Type | Description |
|---|---|
OAuthTokens
|
OAuth tokens from the completed authorization. |
Raises:
| Type | Description |
|---|---|
OAuthRedirectError
|
If the user denies access. |
OAuthRedirectTimeout
|
If the redirect listener times out. |
AuthenticationError
|
If token exchange fails. |
get_access_token
get_access_token() -> str
Return a valid Integration access token, refreshing if needed (sync).
Returns:
| Type | Description |
|---|---|
str
|
Valid bearer access token. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If Integration is not yet authorized. |
aget_access_token
async
aget_access_token() -> str
Return a valid Integration access token, refreshing if needed.
Returns:
| Type | Description |
|---|---|
str
|
Valid bearer access token. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If Integration is not yet authorized. |
webex_byova.auth.service_app.ServiceAppTokenManager
Fetch, refresh, and store per-organization Service App tokens.
Service App tokens are scoped to a customer organization and are obtained after a customer admin authorizes your Service App in Control Hub.
_credentials
instance-attribute
_credentials = credentials
_integration
instance-attribute
_integration = integration
_http
instance-attribute
_http = http
_storage
instance-attribute
_storage = storage
_application_id
instance-attribute
_application_id = derive_application_id(client_id)
application_id
property
application_id: str
Base64url-encoded Service App application ID derived from client ID.
credentials
property
credentials: ServiceAppCredentials
Service App OAuth client credentials.
__init__
__init__(credentials: ServiceAppCredentials, integration: IntegrationTokenManager, http: HttpClient, storage: TokenStorage) -> None
Initialize the Service App token manager.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
credentials
|
ServiceAppCredentials
|
Service App OAuth client credentials. |
required |
integration
|
IntegrationTokenManager
|
Integration token manager for bearer authentication. |
required |
http
|
HttpClient
|
Shared HTTP client for API requests. |
required |
storage
|
TokenStorage
|
Token storage backend. |
required |
_parse_token_response
_parse_token_response(data: dict) -> ServiceAppTokens
afetch_token_for_org
async
afetch_token_for_org(org_id: str) -> ServiceAppTokens
Fetch Service App tokens for an organization using Integration bearer auth.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Target organization UUID. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens
|
Service App tokens, persisted to storage. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If the token fetch fails. |
fetch_token_for_org
fetch_token_for_org(org_id: str) -> ServiceAppTokens
Fetch Service App tokens for an organization (sync wrapper).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Target organization UUID. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens
|
Service App tokens, persisted to storage. |
afetch_token_from_webhook_org
async
afetch_token_from_webhook_org(encoded_org_id: str) -> tuple[str, ServiceAppTokens]
Decode org ID from a webhook payload and fetch Service App tokens.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
encoded_org_id
|
str
|
Base64url-encoded org ID from the webhook |
required |
Returns:
| Type | Description |
|---|---|
tuple[str, ServiceAppTokens]
|
Tuple of |
save_registration
save_registration(org_id: str, refresh_token: str) -> ServiceAppTokens
Store tokens when a refresh token is already available (sync).
Use this sandbox path when you already have a Service App refresh token without waiting for a production webhook.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
refresh_token
|
str
|
Existing Service App refresh token. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens
|
Refreshed Service App tokens. |
asave_registration
async
asave_registration(org_id: str, refresh_token: str) -> ServiceAppTokens
Store tokens when a refresh token is already available.
Use this sandbox path when you already have a Service App refresh token without waiting for a production webhook.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
refresh_token
|
str
|
Existing Service App refresh token. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens
|
Refreshed Service App tokens. |
arefresh_for_org
async
arefresh_for_org(org_id: str) -> ServiceAppTokens
Refresh the Service App access token for an organization.
If no refresh token is stored, falls back to fetching new tokens via the Integration bearer.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens
|
Refreshed Service App tokens, persisted to storage. |
aget_access_token
async
aget_access_token(org_id: str) -> str
Return a valid Service App access token for an organization.
Refreshes automatically if the token is missing or expired.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
Returns:
| Type | Description |
|---|---|
str
|
Valid bearer access token for org-scoped API calls. |
Raises:
| Type | Description |
|---|---|
AuthenticationError
|
If the org is not registered. |
webex_byova.auth.storage.TokenStorage
Bases: Protocol
Protocol for storing Integration and per-org Service App tokens.
Implement this protocol for production persistence (Redis, database, etc.). All methods are async to support non-blocking I/O backends.
get_integration_tokens
async
get_integration_tokens() -> OAuthTokens | None
Return stored Integration OAuth tokens, or None if not authorized.
set_integration_tokens
async
set_integration_tokens(tokens: OAuthTokens) -> None
Persist Integration OAuth tokens.
get_service_app_tokens
async
get_service_app_tokens(org_id: str) -> ServiceAppTokens | None
Return Service App tokens for an org, or None if not registered.
set_service_app_tokens
async
set_service_app_tokens(org_id: str, tokens: ServiceAppTokens) -> None
Persist Service App tokens for an organization.
delete_service_app_tokens
async
delete_service_app_tokens(org_id: str) -> None
Remove Service App tokens for an organization (deauthorization).
list_registered_orgs
async
list_registered_orgs() -> list[str]
Return organization IDs with stored Service App tokens.
webex_byova.auth.storage.InMemoryTokenStorage
Default in-process token storage for development and testing.
Tokens are lost when the process exits. Use a custom TokenStorage
implementation for production deployments.
_integration
instance-attribute
_integration: OAuthTokens | None = None
_service_apps
instance-attribute
_service_apps: dict[str, ServiceAppTokens] = {}
__init__
__init__() -> None
Initialize empty in-memory token stores.
get_integration_tokens
async
get_integration_tokens() -> OAuthTokens | None
Return stored Integration OAuth tokens.
Returns:
| Type | Description |
|---|---|
OAuthTokens | None
|
Integration tokens, or |
set_integration_tokens
async
set_integration_tokens(tokens: OAuthTokens) -> None
Store Integration OAuth tokens.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tokens
|
OAuthTokens
|
Tokens to persist. |
required |
get_service_app_tokens
async
get_service_app_tokens(org_id: str) -> ServiceAppTokens | None
Return Service App tokens for an organization.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
Returns:
| Type | Description |
|---|---|
ServiceAppTokens | None
|
Service App tokens, or |
set_service_app_tokens
async
set_service_app_tokens(org_id: str, tokens: ServiceAppTokens) -> None
Store Service App tokens for an organization.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
tokens
|
ServiceAppTokens
|
Tokens to persist. |
required |
delete_service_app_tokens
async
delete_service_app_tokens(org_id: str) -> None
Remove Service App tokens for an organization.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
org_id
|
str
|
Organization UUID. |
required |
list_registered_orgs
async
list_registered_orgs() -> list[str]
Return all organization IDs with stored Service App tokens.
Returns:
| Type | Description |
|---|---|
list[str]
|
List of registered organization UUIDs. |
webex_byova.auth.credentials.load_credentials_from_env
load_credentials_from_env() -> tuple[IntegrationCredentials, ServiceAppCredentials]
Load Integration and Service App credentials from environment variables.
Required variables
WEBEX_INTEGRATION_CLIENT_IDWEBEX_INTEGRATION_CLIENT_SECRETWEBEX_SA_CLIENT_IDWEBEX_SA_CLIENT_SECRET
Optional variables
WEBEX_INTEGRATION_REDIRECT_URI(defaults tohttp://127.0.0.1:8765/callback)
Returns:
| Type | Description |
|---|---|
tuple[IntegrationCredentials, ServiceAppCredentials]
|
Tuple of |
Raises:
| Type | Description |
|---|---|
ValueError
|
If any required environment variable is missing. |
webex_byova.auth.utils.derive_application_id
derive_application_id(service_app_client_id: str) -> str
Derive the Service App application ID from its client ID.
Webex encodes the application ID as base64url of
ciscospark://us/APPLICATION/{client_id}.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
service_app_client_id
|
str
|
Service App OAuth client ID. |
required |
Returns:
| Type | Description |
|---|---|
str
|
Base64url-encoded application ID used in API paths. |
webex_byova.auth.utils.decode_org_id
decode_org_id(encoded_org_id: str) -> str
Decode an organization ID from a webhook orgId value.
Webex sends a base64url-encoded URI in authorized webhook payloads. This extracts the org UUID from the decoded path.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
encoded_org_id
|
str
|
Base64url-encoded org identifier from a webhook. |
required |
Returns:
| Type | Description |
|---|---|
str
|
Decoded organization UUID, or the full decoded string if parsing fails. |
webex_byova.auth.integration.DEFAULT_INTEGRATION_SCOPES
module-attribute
DEFAULT_INTEGRATION_SCOPES: list[str] = ['spark:all', 'spark:applications_token', 'application:webhooks_write', 'application:webhooks_read']
Default OAuth scopes required for BYOVA Integration authorization.