Implementation:OpenHands OpenHands LinearManager
| Knowledge Sources | |
|---|---|
| Domains | Platform_Integration, Linear |
| Last Updated | 2026-02-11 21:00 GMT |
Overview
Orchestrates the full lifecycle of Linear webhook events including HMAC validation, payload parsing, user authentication, and conversation creation.
Description
LinearManager extends the base Manager class to handle Linear integration webhooks. It verifies HMAC-SHA256 webhook signatures using workspace-specific secrets, parses comment and label-update events, authenticates users via Keycloak correlation, fetches repositories through the Linear GraphQL API, and delegates conversation creation to LinearFactory view objects.
Usage
Use this class when processing incoming Linear webhook events. It is instantiated with a TokenManager and handles the complete flow from webhook reception through conversation creation.
Code Reference
Source Location
- Repository: OpenHands
- File: enterprise/integrations/linear/linear_manager.py
- Lines: 1-532
Signature
class LinearManager(Manager):
def __init__(self, token_manager: TokenManager): ...
async def authenticate_user(
self, linear_user_id: str, workspace_id: int
) -> tuple[LinearUser | None, UserAuth | None]: ...
async def validate_request(
self, request: Request
) -> tuple[bool, Optional[str], Optional[Dict]]: ...
def parse_webhook(self, payload: Dict) -> JobContext | None: ...
async def _query_api(self, query: str, variables: Dict) -> Dict: ...
async def _get_repositories(self, user_auth: UserAuth) -> list[Repository]: ...
Import
from enterprise.integrations.linear.linear_manager import LinearManager
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| token_manager | TokenManager | Yes | Manages authentication tokens |
| request | Request | Yes | FastAPI request object for signature validation |
| payload | Dict | Yes | Linear webhook JSON payload |
| linear_user_id | str | Yes | User ID within Linear workspace |
| workspace_id | int | Yes | Linear workspace database identifier |
Outputs
| Name | Type | Description |
|---|---|---|
| validate_request() | tuple[bool, str, Dict] | (is_valid, error_message, parsed_payload) |
| parse_webhook() | JobContext or None | Parsed job context or None if not actionable |
| authenticate_user() | tuple[LinearUser, UserAuth] | User records or None if not found |
Usage Examples
from enterprise.integrations.linear.linear_manager import LinearManager
manager = LinearManager(token_manager=token_mgr)
# Validate incoming webhook
is_valid, error, payload = await manager.validate_request(request)
if not is_valid:
return HTTPException(status_code=401, detail=error)
# Parse the webhook payload
job_context = manager.parse_webhook(payload)
if job_context is None:
return # Not an actionable event
# Authenticate the Linear user
linear_user, user_auth = await manager.authenticate_user(
linear_user_id=job_context.user_id,
workspace_id=job_context.workspace_id,
)