Add Python SDK support for WorkOS Pipes

[birdcar]

Description

This adds the ability for the Python SDK to get WorkOS Pipes tokens associated with an AuthKit user.

Documentation

Does this require changes to the WorkOS Docs? E.g. the API Reference or code snippets need updates.

[x] Yes

If yes, link a related docs PR and add a docs maintainer as a reviewer. Their approval is required.

[greptile-apps]

Greptile Overview

Greptile Summary

This PR adds Python SDK support for WorkOS Pipes by introducing a new Pipes module that enables retrieval of OAuth access tokens for third-party providers associated with AuthKit users.

Implementation Overview

The implementation follows the established SDK architecture patterns:

• Module Structure: Created Pipes and AsyncPipes classes implementing the PipesModule protocol, consistent with other SDK modules
• Type System: Defined Pydantic models in workos/types/pipes/ including AccessToken, GetAccessTokenSuccessResponse, and GetAccessTokenFailureResponse with proper use of Literal types for discriminated unions
• Client Integration: Integrated into both SyncClient and AsyncClient with lazy initialization pattern
• API Endpoint: Calls POST /data-integrations/{provider}/token with user_id and optional organization_id

Key Features

1. Dual Response Handling: The module intelligently handles both success and failure responses by checking the active field and validating against the appropriate Pydantic model
2. Optional Parameters: Supports optional organization_id parameter for scoping token requests
3. Comprehensive Testing: Includes 6 test cases covering sync/async variants, success/failure scenarios, and parameter handling

Code Quality

• ✅ Follows SDK architectural patterns consistently
• ✅ Proper type annotations with Protocol definitions
• ✅ No logging of sensitive access_token data
• ✅ Comprehensive test coverage with both sync and async tests
• ✅ Clean separation of concerns between module, types, and tests
• ✅ Proper integration with BaseClient protocol

The implementation is production-ready and maintains consistency with the existing codebase.

Confidence Score: 5/5

• This PR is safe to merge - it's a well-structured addition that follows all SDK patterns with no security concerns
• Score of 5 reflects a clean, well-tested implementation: follows established SDK patterns perfectly, includes comprehensive test coverage for both sync and async variants, properly handles success/failure cases, integrates cleanly into existing client architecture, and contains no security vulnerabilities or code quality issues
• No files require special attention - all changes are well-structured and follow existing patterns

Important Files Changed

File Analysis

Filename	Score	Overview
workos/pipes.py	5/5	New Pipes module implementing get_access_token for both sync and async clients. Follows SDK patterns correctly with proper type annotations and Protocol definition.
workos/types/pipes/pipes.py	5/5	Pydantic models for access tokens and responses with proper Literal types. Clean type definitions with optional expiry handling.
tests/test_pipes.py	5/5	Comprehensive test coverage for both sync and async variants, covering success/failure cases and optional parameters.
workos/client.py	5/5	Integrated Pipes module into SyncClient with lazy initialization pattern.
workos/async_client.py	5/5	Integrated AsyncPipes module into AsyncClient with lazy initialization pattern.

Sequence Diagram

sequenceDiagram
    participant Client as Client/AsyncClient
    participant Pipes as Pipes/AsyncPipes
    participant HTTP as SyncHTTPClient/AsyncHTTPClient
    participant API as WorkOS API

    Client->>Pipes: get_access_token(provider, user_id, organization_id?)
    activate Pipes
    
    Pipes->>Pipes: Build json_data dict with user_id
    alt organization_id provided
        Pipes->>Pipes: Add organization_id to json_data
    end
    
    Pipes->>HTTP: request("data-integrations/{provider}/token", POST, json_data)
    activate HTTP
    HTTP->>API: POST /data-integrations/{provider}/token
    activate API
    
    alt Success - Token Retrieved
        API-->>HTTP: {"active": true, "access_token": {...}}
        HTTP-->>Pipes: response dict
        Pipes->>Pipes: Check response.get("active") is True
        Pipes->>Pipes: model_validate as GetAccessTokenSuccessResponse
        Pipes-->>Client: GetAccessTokenSuccessResponse
    else Failure - Not Installed or Needs Reauth
        API-->>HTTP: {"active": false, "error": "not_installed|needs_reauthorization"}
        HTTP-->>Pipes: response dict
        Pipes->>Pipes: Check response.get("active") is True (False)
        Pipes->>Pipes: model_validate as GetAccessTokenFailureResponse
        Pipes-->>Client: GetAccessTokenFailureResponse
    else HTTP Error (4xx/5xx)
        API-->>HTTP: Error response
        HTTP->>HTTP: Raise WorkOS Exception
        HTTP-->>Pipes: Exception
        Pipes-->>Client: Exception
    end
    
    deactivate API
    deactivate HTTP
    deactivate Pipes

Loading

[gjtorikian]

lgtm