Integrations Module

The integrations module provides access to integration configurations and OAuth credentials.

Import

from bifrost import integrations

Methods

integrations.get()

Get integration configuration with OAuth tokens.

async def get(
    name: str,
    scope: str | None = None,
) -> IntegrationData | None

Parameters

Parameter	Type	Description
name	str	Integration name (e.g., “HaloPSA”, “Microsoft_Graph”)
scope	str | None	Organization scope (see below)

Scope parameter:

• None (default): Use execution context’s organization
• Org UUID string: Target specific organization
• "global": Access platform-level integration defaults

Returns

IntegrationData object or None if not found.

Example

integration = await integrations.get("Microsoft_Graph")
if integration:
    # Access OAuth tokens
    token = integration.oauth.access_token

    # Access config values
    tenant_id = integration.config.get("tenant_id")

    # Entity ID for multi-tenant
    entity = integration.entity_id

integrations.list_mappings()

List all organization mappings for an integration.

async def list_mappings(name: str) -> list[IntegrationMappingResponse] | None

Example

mappings = await integrations.list_mappings("HaloPSA")
if mappings:
    for mapping in mappings:
        print(f"Org: {mapping.organization_id}, Entity: {mapping.entity_id}")

integrations.get_mapping()

Get a specific organization mapping for an integration.

async def get_mapping(
    name: str,
    scope: str | None = None,
    entity_id: str | None = None,
) -> IntegrationMappingResponse | None

Parameters

Parameter	Type	Description
name	str	Integration name
scope	str | None	Organization scope (see below)
entity_id	str	External entity ID to filter by

Scope parameter:

• None (default): Use execution context’s organization
• Org UUID string: Target specific organization

Example

# Get by scope
mapping = await integrations.get_mapping("HaloPSA", scope="org-123")

# Get by entity_id
mapping = await integrations.get_mapping("HaloPSA", entity_id="tenant-456")

if mapping:
    print(f"Entity: {mapping.entity_id}, Config: {mapping.config}")

integrations.upsert_mapping()

Create or update an organization mapping for an integration.

async def upsert_mapping(
    name: str,
    scope: str,            # Required
    entity_id: str,
    entity_name: str | None = None,
    config: dict | None = None,
) -> IntegrationMappingResponse

Parameters

Parameter	Type	Description
name	str	Integration name
scope	str	Organization ID (required)
entity_id	str	External entity ID (tenant, company, etc.)
entity_name	str	Display name for the entity
config	dict	Organization-specific configuration overrides

Example

mapping = await integrations.upsert_mapping(
    "HaloPSA",
    scope="org-123",
    entity_id="tenant-456",
    entity_name="Acme Corp",
    config={"custom_field": "value"}
)

integrations.delete_mapping()

Delete an organization mapping for an integration.

async def delete_mapping(
    name: str,
    scope: str,            # Required
) -> bool

Parameters

Parameter	Type	Description
name	str	Integration name
scope	str	Organization ID (required)

Example

deleted = await integrations.delete_mapping("HaloPSA", scope="org-123")

Types

IntegrationData

class IntegrationData(BaseModel):
    integration_id: str
    entity_id: str | None        # External entity ID (tenant, company, etc.)
    entity_name: str | None      # Display name for entity
    config: dict                  # Merged configuration (defaults + org overrides)
    oauth: OAuthCredentials | None

OAuthCredentials

class OAuthCredentials(BaseModel):
    connection_name: str
    client_id: str | None
    client_secret: str | None     # Decrypted
    authorization_url: str | None
    token_url: str | None
    scopes: list[str]
    access_token: str | None      # Decrypted
    refresh_token: str | None     # Decrypted
    expires_at: str | None        # ISO format

IntegrationMappingResponse

class IntegrationMappingResponse(BaseModel):
    id: str
    integration_id: str
    organization_id: str
    entity_id: str | None
    entity_name: str | None
    oauth_token_id: str | None
    config: dict
    created_at: datetime
    updated_at: datetime

Usage Pattern

from bifrost import workflow, integrations
import httpx

@workflow
async def sync_users():
    # Get integration with OAuth
    msft = await integrations.get("Microsoft_Graph")
    if not msft or not msft.oauth:
        return {"error": "Integration not configured"}

    # Use the access token
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://graph.microsoft.com/v1.0/users",
            headers={"Authorization": f"Bearer {msft.oauth.access_token}"}
        )
        return response.json()

Configuration Merging

Integration config is merged from two levels:

1. Integration defaults - Platform-wide defaults
2. Organization overrides - Per-org customizations

The config dict in IntegrationData contains the merged result.

Token Handling

OAuth tokens are:

• Encrypted at rest in the database
• Automatically decrypted when retrieved via SDK
• Refreshed automatically by the platform when expired

See Also

• Integrations Overview - Concept guide
• Creating Integrations - Setup guide