wehub-resource-sync
1696 lines
65 KiB
Python
1696 lines
65 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import functools
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import time
|
|
import traceback
|
|
import typing as t
|
|
import uuid
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from enum import Enum
|
|
from unittest import mock
|
|
|
|
import typing_extensions as te
|
|
from composio_client import APIStatusError, Omit, omit
|
|
from composio_client.types import TriggersTypeRetrieveResponse
|
|
from pysher import Pusher
|
|
from pysher.channel import Channel as PusherChannel
|
|
from pysher.connection import Connection as PusherConnection
|
|
|
|
from composio import exceptions
|
|
from composio.client import HttpClient
|
|
from composio.client.types import trigger_instance_upsert_response
|
|
from composio.core.models.base import Resource
|
|
from composio.core.models.internal import Internal
|
|
from composio.core.types import ToolkitVersionParam
|
|
from composio.exceptions import ComposioSDKTimeoutError
|
|
from composio.utils.logging import WithLogger
|
|
from composio.utils.pydantic import none_to_omit
|
|
|
|
PUSHER_AUTH_URL = "{base_url}/api/v3/internal/sdk/realtime/auth?source=python"
|
|
|
|
"""
|
|
export type _TriggerData = {
|
|
appName: string;
|
|
clientId: number;
|
|
payload: Record<string, unknown>;
|
|
originalPayload: Record<string, unknown>;
|
|
metadata: {
|
|
id: string;
|
|
nanoId: string;
|
|
triggerName: string;
|
|
triggerData: string;
|
|
triggerConfig: Record<string, unknown>;
|
|
connection: {
|
|
id: string;
|
|
connectedAccountNanoId: string;
|
|
integrationId: string;
|
|
authConfigNanoId: string;
|
|
clientUniqueUserId: string;
|
|
status: string;
|
|
};
|
|
};
|
|
};
|
|
"""
|
|
|
|
|
|
class _ConnectionData(te.TypedDict):
|
|
id: str
|
|
status: t.Literal["ACTIVE", "INACTIVE"]
|
|
integrationId: str
|
|
authConfigNanoId: str
|
|
clientUniqueUserId: str
|
|
connectedAccountNanoId: str
|
|
|
|
|
|
class _TriggerMetadata(te.TypedDict):
|
|
id: str
|
|
nanoId: str
|
|
triggerName: str
|
|
triggerData: str
|
|
triggerConfig: t.Dict
|
|
connection: _ConnectionData
|
|
|
|
|
|
class _TriggerData(te.TypedDict):
|
|
clientId: str
|
|
appName: str
|
|
payload: t.Dict
|
|
metadata: _TriggerMetadata
|
|
originalPayload: t.Dict
|
|
|
|
|
|
class WebhookVersion(str, Enum):
|
|
"""Webhook payload version."""
|
|
|
|
V1 = "V1"
|
|
V2 = "V2"
|
|
V3 = "V3"
|
|
|
|
|
|
class WebhookPayloadV1(te.TypedDict):
|
|
"""V1 webhook payload structure."""
|
|
|
|
trigger_name: str
|
|
connection_id: str
|
|
trigger_id: str
|
|
payload: t.Dict[str, t.Any]
|
|
log_id: str
|
|
|
|
|
|
class WebhookPayloadV2(te.TypedDict):
|
|
"""V2 webhook payload structure."""
|
|
|
|
type: str
|
|
timestamp: str
|
|
log_id: str
|
|
data: t.Dict[str, t.Any]
|
|
|
|
|
|
class WebhookPayloadV3(te.TypedDict):
|
|
"""V3 webhook payload structure - generic envelope for all composio.* events."""
|
|
|
|
id: str
|
|
timestamp: str
|
|
type: str # Any composio.* event type (e.g., 'composio.trigger.message', 'composio.connected_account.expired')
|
|
metadata: t.Dict[
|
|
str, t.Any
|
|
] # Shape varies by event type (trigger vs connection events)
|
|
data: t.Dict[str, t.Any]
|
|
|
|
|
|
class WebhookTriggerPayloadV3Metadata(te.TypedDict):
|
|
"""V3 trigger-specific webhook payload metadata."""
|
|
|
|
log_id: str
|
|
trigger_slug: str
|
|
trigger_id: str
|
|
connected_account_id: str
|
|
auth_config_id: str
|
|
user_id: str
|
|
|
|
|
|
class WebhookTriggerPayloadV3(te.TypedDict):
|
|
"""V3 trigger-specific webhook payload structure."""
|
|
|
|
id: str
|
|
timestamp: str
|
|
type: str
|
|
metadata: WebhookTriggerPayloadV3Metadata
|
|
data: t.Dict[str, t.Any]
|
|
|
|
|
|
WebhookPayload = t.Union[WebhookPayloadV1, WebhookPayloadV2, WebhookPayloadV3]
|
|
|
|
|
|
class TriggerConnectedAccountSchema(t.TypedDict):
|
|
id: str
|
|
uuid: str
|
|
user_id: str
|
|
auth_config_id: str
|
|
auth_config_uuid: str
|
|
status: t.Literal["ACTIVE", "INACTIVE"]
|
|
|
|
|
|
class TriggerMetadataSchema(t.TypedDict):
|
|
id: str
|
|
toolkit_slug: str
|
|
trigger_slug: str
|
|
trigger_data: t.Optional[str]
|
|
trigger_config: t.Dict[str, t.Any]
|
|
connected_account: TriggerConnectedAccountSchema
|
|
|
|
|
|
class TriggerEvent(t.TypedDict):
|
|
id: str # The ID of the trigger
|
|
uuid: str # UUID of the trigger
|
|
user_id: str # The ID of the user that triggered the event
|
|
|
|
trigger_slug: str # The slug of the trigger that triggered the event
|
|
toolkit_slug: str # The slug of the toolkit that triggered the event
|
|
|
|
payload: t.Optional[t.Dict[str, t.Any]] # The payload of the trigger
|
|
metadata: TriggerMetadataSchema
|
|
original_payload: t.Optional[
|
|
t.Dict[str, t.Any]
|
|
] # The original payload of the trigger
|
|
|
|
|
|
class VerifyWebhookResult(t.TypedDict):
|
|
"""Result of webhook verification."""
|
|
|
|
version: WebhookVersion # The webhook version (V1, V2, or V3)
|
|
payload: TriggerEvent # The parsed and normalized webhook payload
|
|
raw_payload: WebhookPayload # The original parsed payload
|
|
|
|
|
|
class WebhookSubscription(t.TypedDict, total=False):
|
|
"""Webhook subscription returned by the Composio API."""
|
|
|
|
id: str
|
|
webhook_url: str
|
|
version: str
|
|
enabled_events: t.List[str]
|
|
secret: str
|
|
created_at: str
|
|
updated_at: str
|
|
|
|
|
|
DEFAULT_WEBHOOK_SUBSCRIPTION_EVENTS = ("composio.trigger.message",)
|
|
WEBHOOK_SUBSCRIPTIONS_PATH = "/api/v3.1/webhook_subscriptions"
|
|
|
|
|
|
_ = {
|
|
"appName": "github",
|
|
"payload": {
|
|
"author": "angrybayblade",
|
|
"id": "a2334682759c4324e911d8f52f8fb6bdf1338d94",
|
|
"message": "temp",
|
|
"timestamp": "2025-06-19T11:28:01+05:30",
|
|
"url": "https://github.com/angrybayblade/ph7/commit/a2334682759c4324e911d8f52f8fb6bdf1338d94",
|
|
},
|
|
"originalPayload": {
|
|
"ref": "refs/heads/temp",
|
|
"before": "e5eb72bc9f973b6cc82e0de9708660c5341befe5",
|
|
"after": "a2334682759c4324e911d8f52f8fb6bdf1338d94",
|
|
"repository": {
|
|
"id": 752122629,
|
|
"node_id": "R_kgDOLNR7BQ",
|
|
"name": "ph7",
|
|
"full_name": "angrybayblade/ph7",
|
|
"private": False,
|
|
"owner": {
|
|
"name": "angrybayblade",
|
|
"email": "35092918+angrybayblade@users.noreply.github.com",
|
|
"login": "angrybayblade",
|
|
"id": 35092918,
|
|
"node_id": "MDQ6VXNlcjM1MDkyOTE4",
|
|
"avatar_url": "https://avatars.githubusercontent.com/u/35092918?v=4",
|
|
"gravatar_id": "",
|
|
"url": "https://api.github.com/users/angrybayblade",
|
|
"html_url": "https://github.com/angrybayblade",
|
|
"followers_url": "https://api.github.com/users/angrybayblade/followers",
|
|
"following_url": "https://api.github.com/users/angrybayblade/following{/other_user}",
|
|
"gists_url": "https://api.github.com/users/angrybayblade/gists{/gist_id}",
|
|
"starred_url": "https://api.github.com/users/angrybayblade/starred{/owner}{/repo}",
|
|
"subscriptions_url": "https://api.github.com/users/angrybayblade/subscriptions",
|
|
"organizations_url": "https://api.github.com/users/angrybayblade/orgs",
|
|
"repos_url": "https://api.github.com/users/angrybayblade/repos",
|
|
"events_url": "https://api.github.com/users/angrybayblade/events{/privacy}",
|
|
"received_events_url": "https://api.github.com/users/angrybayblade/received_events",
|
|
"type": "User",
|
|
"user_view_type": "public",
|
|
"site_admin": False,
|
|
},
|
|
"html_url": "https://github.com/angrybayblade/ph7",
|
|
"description": "💧 Python native HTML rendering",
|
|
"fork": False,
|
|
"url": "https://api.github.com/repos/angrybayblade/ph7",
|
|
"forks_url": "https://api.github.com/repos/angrybayblade/ph7/forks",
|
|
"keys_url": "https://api.github.com/repos/angrybayblade/ph7/keys{/key_id}",
|
|
"collaborators_url": "https://api.github.com/repos/angrybayblade/ph7/collaborators{/collaborator}",
|
|
"teams_url": "https://api.github.com/repos/angrybayblade/ph7/teams",
|
|
"hooks_url": "https://api.github.com/repos/angrybayblade/ph7/hooks",
|
|
"issue_events_url": "https://api.github.com/repos/angrybayblade/ph7/issues/events{/number}",
|
|
"events_url": "https://api.github.com/repos/angrybayblade/ph7/events",
|
|
"assignees_url": "https://api.github.com/repos/angrybayblade/ph7/assignees{/user}",
|
|
"branches_url": "https://api.github.com/repos/angrybayblade/ph7/branches{/branch}",
|
|
"tags_url": "https://api.github.com/repos/angrybayblade/ph7/tags",
|
|
"blobs_url": "https://api.github.com/repos/angrybayblade/ph7/git/blobs{/sha}",
|
|
"git_tags_url": "https://api.github.com/repos/angrybayblade/ph7/git/tags{/sha}",
|
|
"git_refs_url": "https://api.github.com/repos/angrybayblade/ph7/git/refs{/sha}",
|
|
"trees_url": "https://api.github.com/repos/angrybayblade/ph7/git/trees{/sha}",
|
|
"statuses_url": "https://api.github.com/repos/angrybayblade/ph7/statuses/{sha}",
|
|
"languages_url": "https://api.github.com/repos/angrybayblade/ph7/languages",
|
|
"stargazers_url": "https://api.github.com/repos/angrybayblade/ph7/stargazers",
|
|
"contributors_url": "https://api.github.com/repos/angrybayblade/ph7/contributors",
|
|
"subscribers_url": "https://api.github.com/repos/angrybayblade/ph7/subscribers",
|
|
"subscription_url": "https://api.github.com/repos/angrybayblade/ph7/subscription",
|
|
"commits_url": "https://api.github.com/repos/angrybayblade/ph7/commits{/sha}",
|
|
"git_commits_url": "https://api.github.com/repos/angrybayblade/ph7/git/commits{/sha}",
|
|
"comments_url": "https://api.github.com/repos/angrybayblade/ph7/comments{/number}",
|
|
"issue_comment_url": "https://api.github.com/repos/angrybayblade/ph7/issues/comments{/number}",
|
|
"contents_url": "https://api.github.com/repos/angrybayblade/ph7/contents/{+path}",
|
|
"compare_url": "https://api.github.com/repos/angrybayblade/ph7/compare/{base}...{head}",
|
|
"merges_url": "https://api.github.com/repos/angrybayblade/ph7/merges",
|
|
"archive_url": "https://api.github.com/repos/angrybayblade/ph7/{archive_format}{/ref}",
|
|
"downloads_url": "https://api.github.com/repos/angrybayblade/ph7/downloads",
|
|
"issues_url": "https://api.github.com/repos/angrybayblade/ph7/issues{/number}",
|
|
"pulls_url": "https://api.github.com/repos/angrybayblade/ph7/pulls{/number}",
|
|
"milestones_url": "https://api.github.com/repos/angrybayblade/ph7/milestones{/number}",
|
|
"notifications_url": "https://api.github.com/repos/angrybayblade/ph7/notifications{?since,all,participating}",
|
|
"labels_url": "https://api.github.com/repos/angrybayblade/ph7/labels{/name}",
|
|
"releases_url": "https://api.github.com/repos/angrybayblade/ph7/releases{/id}",
|
|
"deployments_url": "https://api.github.com/repos/angrybayblade/ph7/deployments",
|
|
"created_at": 1706936166,
|
|
"updated_at": "2025-03-22T17:26:27Z",
|
|
"pushed_at": 1750312684,
|
|
"git_url": "git://github.com/angrybayblade/ph7.git",
|
|
"ssh_url": "git@github.com:angrybayblade/ph7.git",
|
|
"clone_url": "https://github.com/angrybayblade/ph7.git",
|
|
"svn_url": "https://github.com/angrybayblade/ph7",
|
|
"homepage": "https://angrybayblade.github.io/ph7/",
|
|
"size": 3031,
|
|
"stargazers_count": 7,
|
|
"watchers_count": 7,
|
|
"language": "Python",
|
|
"has_issues": True,
|
|
"has_projects": True,
|
|
"has_downloads": True,
|
|
"has_wiki": True,
|
|
"has_pages": True,
|
|
"has_discussions": False,
|
|
"forks_count": 1,
|
|
"mirror_url": None,
|
|
"archived": False,
|
|
"disabled": False,
|
|
"open_issues_count": 0,
|
|
"license": None,
|
|
"allow_forking": True,
|
|
"is_template": False,
|
|
"web_commit_signoff_required": False,
|
|
"topics": [
|
|
"css",
|
|
"django",
|
|
"flask",
|
|
"html",
|
|
"js",
|
|
"template-engine",
|
|
"web",
|
|
],
|
|
"visibility": "public",
|
|
"forks": 1,
|
|
"open_issues": 0,
|
|
"watchers": 7,
|
|
"default_branch": "main",
|
|
"stargazers": 7,
|
|
"master_branch": "main",
|
|
},
|
|
"pusher": {
|
|
"name": "angrybayblade",
|
|
"email": "35092918+angrybayblade@users.noreply.github.com",
|
|
},
|
|
"sender": {
|
|
"login": "angrybayblade",
|
|
"id": 35092918,
|
|
"node_id": "MDQ6VXNlcjM1MDkyOTE4",
|
|
"avatar_url": "https://avatars.githubusercontent.com/u/35092918?v=4",
|
|
"gravatar_id": "",
|
|
"url": "https://api.github.com/users/angrybayblade",
|
|
"html_url": "https://github.com/angrybayblade",
|
|
"followers_url": "https://api.github.com/users/angrybayblade/followers",
|
|
"following_url": "https://api.github.com/users/angrybayblade/following{/other_user}",
|
|
"gists_url": "https://api.github.com/users/angrybayblade/gists{/gist_id}",
|
|
"starred_url": "https://api.github.com/users/angrybayblade/starred{/owner}{/repo}",
|
|
"subscriptions_url": "https://api.github.com/users/angrybayblade/subscriptions",
|
|
"organizations_url": "https://api.github.com/users/angrybayblade/orgs",
|
|
"repos_url": "https://api.github.com/users/angrybayblade/repos",
|
|
"events_url": "https://api.github.com/users/angrybayblade/events{/privacy}",
|
|
"received_events_url": "https://api.github.com/users/angrybayblade/received_events",
|
|
"type": "User",
|
|
"user_view_type": "public",
|
|
"site_admin": False,
|
|
},
|
|
"created": False,
|
|
"deleted": False,
|
|
"forced": False,
|
|
"base_ref": None,
|
|
"compare": "https://github.com/angrybayblade/ph7/compare/e5eb72bc9f97...a2334682759c",
|
|
"commits": [
|
|
{
|
|
"id": "a2334682759c4324e911d8f52f8fb6bdf1338d94",
|
|
"tree_id": "ed0ae8c902cb2a0e12d3c64aea8aab66f7b06054",
|
|
"distinct": True,
|
|
"message": "temp",
|
|
"timestamp": "2025-06-19T11:28:01+05:30",
|
|
"url": "https://github.com/angrybayblade/ph7/commit/a2334682759c4324e911d8f52f8fb6bdf1338d94",
|
|
"author": {
|
|
"name": "angrybayblade",
|
|
"email": "vptl185@gmail.com",
|
|
"username": "angrybayblade",
|
|
},
|
|
"committer": {
|
|
"name": "angrybayblade",
|
|
"email": "vptl185@gmail.com",
|
|
"username": "angrybayblade",
|
|
},
|
|
"added": [],
|
|
"removed": [],
|
|
"modified": ["mkdocs.yml"],
|
|
}
|
|
],
|
|
"head_commit": {
|
|
"id": "a2334682759c4324e911d8f52f8fb6bdf1338d94",
|
|
"tree_id": "ed0ae8c902cb2a0e12d3c64aea8aab66f7b06054",
|
|
"distinct": True,
|
|
"message": "temp",
|
|
"timestamp": "2025-06-19T11:28:01+05:30",
|
|
"url": "https://github.com/angrybayblade/ph7/commit/a2334682759c4324e911d8f52f8fb6bdf1338d94",
|
|
"author": {
|
|
"name": "angrybayblade",
|
|
"email": "vptl185@gmail.com",
|
|
"username": "angrybayblade",
|
|
},
|
|
"committer": {
|
|
"name": "angrybayblade",
|
|
"email": "vptl185@gmail.com",
|
|
"username": "angrybayblade",
|
|
},
|
|
"added": [],
|
|
"removed": [],
|
|
"modified": ["mkdocs.yml"],
|
|
},
|
|
},
|
|
"metadata": {
|
|
"id": "2507236e-4be2-4606-b338-fb24c76ce38d",
|
|
"nanoId": "ti_VZEWUqF5fQ6P",
|
|
"connectionId": "75271255-fe95-4d34-bee3-459a02645e38",
|
|
"connectionNanoId": "ca_5KdA-e2C4ZMd",
|
|
"triggerName": "GITHUB_COMMIT_EVENT",
|
|
"triggerData": '{"event_type": "push", "github_hook_id": "552965247"}',
|
|
"triggerConfig": {"repo": "ph7", "owner": "angrybayblade"},
|
|
"connection": {
|
|
"id": "75271255-fe95-4d34-bee3-459a02645e38",
|
|
"connectedAccountNanoId": "ca_5KdA-e2C4ZMd",
|
|
"integrationId": "500748ea-8547-4abb-9f9e-10dbcfdb81c2",
|
|
"authConfigNanoId": "ac_ZxnpxqOo1nAP",
|
|
"clientUniqueUserId": "default",
|
|
"status": "ACTIVE",
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
class _ChunkedTriggerEventData(te.TypedDict):
|
|
"""Cunked trigger event data model."""
|
|
|
|
id: str
|
|
index: int
|
|
chunk: str
|
|
final: bool
|
|
|
|
|
|
class TriggerEventFilters(te.TypedDict):
|
|
"""Trigger event filterset."""
|
|
|
|
trigger_slug: te.NotRequired[str]
|
|
trigger_id: te.NotRequired[str]
|
|
toolkit: te.NotRequired[str]
|
|
user_id: te.NotRequired[str]
|
|
auth_config_id: te.NotRequired[str]
|
|
connected_account_id: te.NotRequired[str]
|
|
|
|
|
|
TriggerCallback = t.Callable[[TriggerEvent], None]
|
|
|
|
|
|
# Realtime trigger frames can carry message bodies / PII, so the raw frame is
|
|
# never logged in full — only a bounded preview when it fails to parse.
|
|
_MAX_LOGGED_FRAME_CHARS = 512
|
|
|
|
|
|
def _truncate_frame(event: str) -> str:
|
|
"""Return a bounded preview of a raw frame for safe logging."""
|
|
if len(event) <= _MAX_LOGGED_FRAME_CHARS:
|
|
return event
|
|
return f"{event[:_MAX_LOGGED_FRAME_CHARS]}… ({len(event)} chars total)"
|
|
|
|
|
|
def _coerce_str(value: t.Any) -> str:
|
|
"""Coerce a possibly-missing or non-string field to ``str``.
|
|
|
|
Mirrors the TS SDK's ``toStringOrDefault`` so realtime/webhook frames that
|
|
carry ``None`` or a non-string identity field don't later raise
|
|
``AttributeError`` on ``.lower()`` / ``.split()``. ``None`` maps to "".
|
|
"""
|
|
return "" if value is None else str(value)
|
|
|
|
|
|
def _is_v3_envelope(data: t.Any) -> bool:
|
|
"""Return ``True`` if ``data`` is a modern V3 envelope.
|
|
|
|
V3 payloads carry a ``type`` starting with ``composio.`` and a ``metadata``
|
|
object, alongside top-level ``id`` and ``data`` keys. The same envelope is
|
|
delivered over both the webhook channel and the realtime (Pusher) channel,
|
|
so this is used to route either source through the V3 normalizer.
|
|
"""
|
|
if not isinstance(data, dict):
|
|
return False
|
|
event_type = data.get("type", "")
|
|
return (
|
|
isinstance(event_type, str)
|
|
and event_type.startswith("composio.")
|
|
and isinstance(data.get("metadata"), dict)
|
|
and "id" in data
|
|
and "data" in data
|
|
)
|
|
|
|
|
|
def _build_trigger_event_from_v3(data: WebhookPayloadV3) -> TriggerEvent:
|
|
"""Normalize a V3 envelope (webhook or realtime) into a ``TriggerEvent``.
|
|
|
|
Trigger frames are identified by their envelope ``type``
|
|
(``composio.trigger.message``), not by the presence of every metadata
|
|
field. The trigger-specific metadata fields are then read best-effort and
|
|
coerced to ``str`` so a frame that omits or mistypes one of them is still
|
|
delivered rather than silently demoted to a non-trigger event or raising
|
|
``AttributeError`` on the toolkit-slug split.
|
|
"""
|
|
metadata = data.get("metadata") or {}
|
|
event_type = data.get("type", "")
|
|
|
|
if event_type == "composio.trigger.message":
|
|
trigger_id = _coerce_str(metadata.get("trigger_id"))
|
|
trigger_slug = _coerce_str(metadata.get("trigger_slug"))
|
|
user_id = _coerce_str(metadata.get("user_id"))
|
|
connected_account_id = _coerce_str(metadata.get("connected_account_id"))
|
|
auth_config_id = _coerce_str(metadata.get("auth_config_id"))
|
|
toolkit_slug = (
|
|
trigger_slug.split("_")[0].upper() if "_" in trigger_slug else "UNKNOWN"
|
|
)
|
|
return t.cast(
|
|
TriggerEvent,
|
|
{
|
|
"id": trigger_id,
|
|
"uuid": trigger_id,
|
|
"user_id": user_id,
|
|
"toolkit_slug": toolkit_slug,
|
|
"trigger_slug": trigger_slug,
|
|
"metadata": {
|
|
"id": trigger_id,
|
|
"uuid": trigger_id,
|
|
"toolkit_slug": toolkit_slug,
|
|
"trigger_slug": trigger_slug,
|
|
"trigger_data": None,
|
|
"trigger_config": {},
|
|
"connected_account": {
|
|
"id": connected_account_id,
|
|
"uuid": connected_account_id,
|
|
"auth_config_id": auth_config_id,
|
|
"auth_config_uuid": auth_config_id,
|
|
"user_id": user_id,
|
|
"status": "ACTIVE",
|
|
},
|
|
},
|
|
"payload": data.get("data", {}),
|
|
"original_payload": None,
|
|
},
|
|
)
|
|
|
|
# Non-trigger V3 event (e.g., connection expired)
|
|
event_id = _coerce_str(data.get("id"))
|
|
return t.cast(
|
|
TriggerEvent,
|
|
{
|
|
"id": event_id,
|
|
"uuid": event_id,
|
|
"user_id": "",
|
|
"toolkit_slug": "COMPOSIO",
|
|
"trigger_slug": event_type,
|
|
"metadata": {
|
|
"id": event_id,
|
|
"uuid": event_id,
|
|
"toolkit_slug": "COMPOSIO",
|
|
"trigger_slug": event_type,
|
|
"trigger_data": None,
|
|
"trigger_config": {},
|
|
"connected_account": {
|
|
"id": "",
|
|
"uuid": "",
|
|
"auth_config_id": "",
|
|
"auth_config_uuid": "",
|
|
"user_id": "",
|
|
"status": "ACTIVE",
|
|
},
|
|
},
|
|
"payload": data.get("data", {}),
|
|
"original_payload": data,
|
|
},
|
|
)
|
|
|
|
|
|
class TriggerSubscription(Resource):
|
|
"""Trigger subscription."""
|
|
|
|
_pusher: Pusher
|
|
_channel: PusherChannel
|
|
_connection: PusherConnection
|
|
_alive: bool
|
|
|
|
def __init__(self, client: HttpClient) -> None:
|
|
"""Initialize subscription object."""
|
|
super().__init__(client=client)
|
|
self.client = client
|
|
self._alive = False
|
|
self._chunks: t.Dict[str, t.Dict[int, str]] = {}
|
|
self._callbacks: t.List[t.Tuple[TriggerCallback, TriggerEventFilters]] = []
|
|
|
|
def handle(
|
|
self, **filters: te.Unpack[TriggerEventFilters]
|
|
) -> t.Callable[[TriggerCallback], TriggerCallback]:
|
|
"""Register a trigger callaback."""
|
|
|
|
def _wrap(f: TriggerCallback) -> TriggerCallback:
|
|
self.logger.debug(f"Registering callback `{f.__name__}`")
|
|
self._callbacks.append((f, filters))
|
|
return f
|
|
|
|
return _wrap
|
|
|
|
def _parse_payload(self, event: str) -> t.Optional[TriggerEvent]:
|
|
"""Parse a realtime event payload into a ``TriggerEvent``.
|
|
|
|
The realtime (Pusher) channel delivers either the modern V3 envelope
|
|
(for projects on webhook version V3) or the legacy envelope (for V1/V2
|
|
projects). Both shapes are handled here. Any unrecognized or malformed
|
|
frame is logged and skipped (returns ``None``) rather than raising, so a
|
|
single bad frame cannot tear down the subscription.
|
|
"""
|
|
try:
|
|
data = json.loads(event)
|
|
except Exception as e:
|
|
self.logger.warning(f"Error decoding payload: {e}")
|
|
return None
|
|
|
|
try:
|
|
# V3 envelope — same shape as the V3 webhook payload.
|
|
if _is_v3_envelope(data):
|
|
return _build_trigger_event_from_v3(t.cast(WebhookPayloadV3, data))
|
|
|
|
# Legacy envelope (V1/V2 projects).
|
|
legacy = t.cast(_TriggerData, data)
|
|
return t.cast(
|
|
TriggerEvent,
|
|
{
|
|
"id": legacy["metadata"]["nanoId"],
|
|
"uuid": legacy["metadata"]["id"],
|
|
"user_id": legacy["metadata"]["connection"]["clientUniqueUserId"],
|
|
"toolkit_slug": legacy["appName"],
|
|
"trigger_slug": legacy["metadata"]["triggerName"],
|
|
"metadata": {
|
|
"id": legacy["metadata"]["nanoId"],
|
|
"uuid": legacy["metadata"]["id"],
|
|
"toolkit_slug": legacy["appName"],
|
|
"trigger_slug": legacy["metadata"]["triggerName"],
|
|
"trigger_data": legacy["metadata"].get("triggerData"),
|
|
"trigger_config": legacy["metadata"]["triggerConfig"],
|
|
"connected_account": {
|
|
"id": legacy["metadata"]["connection"][
|
|
"connectedAccountNanoId"
|
|
],
|
|
"uuid": legacy["metadata"]["connection"]["id"],
|
|
"auth_config_id": legacy["metadata"]["connection"][
|
|
"authConfigNanoId"
|
|
],
|
|
"auth_config_uuid": legacy["metadata"]["connection"][
|
|
"integrationId"
|
|
],
|
|
"user_id": legacy["metadata"]["connection"][
|
|
"clientUniqueUserId"
|
|
],
|
|
"status": legacy["metadata"]["connection"]["status"],
|
|
},
|
|
},
|
|
"payload": legacy.get("payload"),
|
|
"original_payload": legacy.get("originalPayload"),
|
|
},
|
|
)
|
|
except Exception as e:
|
|
# A single malformed frame must never propagate into pysher's
|
|
# dispatch loop (which calls callbacks with no try/except) and tear
|
|
# down the subscription, so catch broadly — AttributeError and
|
|
# others, not just KeyError/TypeError — and skip the frame.
|
|
self.logger.warning(f"Error parsing trigger payload: {e}")
|
|
return None
|
|
|
|
def _handle_chunked_events(self, event: str) -> None:
|
|
"""Handle chunked events."""
|
|
data = _ChunkedTriggerEventData(**json.loads(event)) # type: ignore
|
|
if data["id"] not in self._chunks:
|
|
self._chunks[data["id"]] = {}
|
|
|
|
self._chunks[data["id"]][data["index"]] = data["chunk"]
|
|
if data["final"]:
|
|
_chunks = self._chunks.pop(data["id"])
|
|
self._handle_event(event="".join([_chunks[idx] for idx in sorted(_chunks)]))
|
|
|
|
def _filters_match(
|
|
self,
|
|
data: TriggerEvent,
|
|
filters: TriggerEventFilters,
|
|
callback: str,
|
|
) -> bool:
|
|
"""Check if filters match the event data."""
|
|
checks = (
|
|
("trigger_slug", data["trigger_slug"]),
|
|
("trigger_id", data["metadata"]["id"]),
|
|
("toolkit", data["toolkit_slug"]),
|
|
("user_id", data["user_id"]),
|
|
("auth_config_id", data["metadata"]["connected_account"]["auth_config_id"]),
|
|
("connected_account_id", data["metadata"]["connected_account"]["id"]),
|
|
)
|
|
for name, check in checks:
|
|
value = filters.get(name)
|
|
if value is None:
|
|
# No filter set for this field — nothing to match against.
|
|
continue
|
|
|
|
# ``check`` is str()-wrapped because realtime frames may carry
|
|
# non-string identity fields; an empty event-side value is treated
|
|
# as a non-match so synthesized "" fields don't fail open into a
|
|
# filter that happens to be "".
|
|
check_str = str(check)
|
|
if check_str and str(value).lower() == check_str.lower():
|
|
continue
|
|
|
|
self.logger.debug(
|
|
f"Skipping `{callback}` since "
|
|
f"`{name}` filter does not match the event metadata",
|
|
)
|
|
return False
|
|
return True
|
|
|
|
def _handle_callback(
|
|
self,
|
|
callback: TriggerCallback,
|
|
data: TriggerEvent,
|
|
filters: TriggerEventFilters,
|
|
) -> t.Any:
|
|
"""Handle callback."""
|
|
if not self._filters_match(data, filters, callback.__name__):
|
|
return
|
|
|
|
try:
|
|
callback(data)
|
|
except Exception:
|
|
self.logger.error(
|
|
f"Error executing `{callback.__name__}` for "
|
|
f"event `{data['metadata']['trigger_slug']}` "
|
|
f"with error:\n {traceback.format_exc()}"
|
|
)
|
|
|
|
def _handle_event(self, event: str) -> None:
|
|
"""Filter events and call the callback function."""
|
|
data = self._parse_payload(event=event)
|
|
if data is None:
|
|
# _parse_payload already logged the specific reason; here we only
|
|
# note that the frame was skipped, with a bounded preview so a
|
|
# PII-bearing frame isn't dumped in full.
|
|
self.logger.error(
|
|
f"Skipping unparseable trigger frame: {_truncate_frame(event)}"
|
|
)
|
|
return
|
|
|
|
self.logger.debug(
|
|
f"Received trigger event with trigger ID: {data['metadata']['id']} "
|
|
f"and trigger name: {data['metadata']['trigger_slug']}"
|
|
)
|
|
awaitables: t.List = []
|
|
with ThreadPoolExecutor() as executor:
|
|
for callback, filters in self._callbacks:
|
|
awaitables.append(
|
|
executor.submit(
|
|
self._handle_callback,
|
|
callback,
|
|
data,
|
|
filters,
|
|
)
|
|
)
|
|
_ = [future.result() for future in awaitables]
|
|
|
|
def is_alive(self) -> bool:
|
|
"""Check if subscription is live."""
|
|
return self._alive
|
|
|
|
def has_errored(self) -> bool:
|
|
"""Check if the connection errored and disconnected."""
|
|
return self._connection.socket is None or self._connection.socket.has_errored
|
|
|
|
def set_alive(self) -> None:
|
|
"""Set `_alive` to True."""
|
|
self._alive = True
|
|
|
|
def wait_forever(self) -> None:
|
|
"""Wait infinitely."""
|
|
while self.is_alive() and not self.has_errored():
|
|
time.sleep(1)
|
|
|
|
def stop(self) -> None:
|
|
"""Stop the trigger listener."""
|
|
self._connection.disconnect()
|
|
self._alive = False
|
|
|
|
def restart(self) -> None:
|
|
"""Restart the subscription connection"""
|
|
self._connection.disconnect()
|
|
self._connection._connect() # pylint: disable=protected-access
|
|
|
|
|
|
class _SubcriptionBuilder(WithLogger):
|
|
"""Pusher client for Composio SDK."""
|
|
|
|
def __init__(self, client: HttpClient) -> None:
|
|
"""Initialize pusher client."""
|
|
super().__init__()
|
|
self._client = client
|
|
self.api_key = self._client.api_key
|
|
self.base_url = self._client.base_url
|
|
self.internal = Internal(client=self._client)
|
|
self.subscription = TriggerSubscription(client=self._client)
|
|
|
|
def _get_connection_handler(
|
|
self,
|
|
project_id: str,
|
|
pusher: Pusher,
|
|
subscription: TriggerSubscription,
|
|
) -> t.Callable[[str], None]:
|
|
def _connection_handler(_: str) -> None:
|
|
channel = t.cast(
|
|
PusherChannel,
|
|
pusher.subscribe(
|
|
channel_name=f"private-{project_id}_triggers",
|
|
),
|
|
)
|
|
channel.bind(
|
|
event_name="trigger_to_client",
|
|
callback=subscription._handle_event,
|
|
)
|
|
channel.bind(
|
|
event_name="chunked-trigger_to_client",
|
|
callback=subscription._handle_chunked_events,
|
|
)
|
|
subscription.set_alive()
|
|
subscription._channel = channel # pylint: disable=protected-access
|
|
subscription._connection = ( # pylint: disable=protected-access
|
|
channel.connection
|
|
)
|
|
|
|
return _connection_handler
|
|
|
|
def _get_pusher_instance(self, key: str, cluster: str) -> Pusher:
|
|
"""Get a pusher instance."""
|
|
return Pusher(
|
|
key=key,
|
|
cluster=cluster,
|
|
auth_endpoint=PUSHER_AUTH_URL.format(base_url=self._client.base_url),
|
|
auth_endpoint_headers={
|
|
"x-api-key": self._client.api_key,
|
|
"x-request-id": str(uuid.uuid4()),
|
|
},
|
|
auto_sub=True,
|
|
)
|
|
|
|
def connect(self, timeout: float = 15.0) -> TriggerSubscription:
|
|
"""Connect to Pusher channel for given client ID."""
|
|
self.logger.debug("Creating trigger subscription")
|
|
project_info = self.internal.get_sdk_realtime_credentials()
|
|
pusher = self._get_pusher_instance(
|
|
key=project_info.pusher_key,
|
|
cluster=project_info.pusher_cluster,
|
|
)
|
|
|
|
# Patch pusher logger
|
|
pusher.connection.logger = mock.MagicMock() # type: ignore
|
|
pusher.connection.bind(
|
|
"pusher:connection_established",
|
|
self._get_connection_handler(
|
|
project_id=project_info.project_id,
|
|
pusher=pusher,
|
|
subscription=self.subscription,
|
|
),
|
|
)
|
|
pusher.connect()
|
|
|
|
# Wait for connection to get established
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
if not self.subscription.is_alive():
|
|
time.sleep(0.5)
|
|
continue
|
|
|
|
self.subscription._pusher = pusher # pylint: disable=protected-access
|
|
return self.subscription
|
|
raise ComposioSDKTimeoutError(
|
|
"Timed out while waiting for trigger listener to be established"
|
|
)
|
|
|
|
|
|
class Triggers(Resource):
|
|
"""Triggers (instance) class"""
|
|
|
|
enable: t.Callable
|
|
"""Enables a trigger given its id"""
|
|
|
|
disable: t.Callable
|
|
"""Disables a trigger given its id"""
|
|
|
|
def __init__(
|
|
self,
|
|
client: HttpClient,
|
|
toolkit_versions: t.Optional[ToolkitVersionParam] = None,
|
|
):
|
|
"""
|
|
Initialize the triggers resource.
|
|
|
|
:param client: The client to use for the triggers resource.
|
|
:param toolkit_versions: The versions of the toolkits to use. Defaults to 'latest' if not provided.
|
|
"""
|
|
self._client = client
|
|
self._toolkit_versions = toolkit_versions
|
|
self.list_enum = self._client.triggers_types.retrieve_enum
|
|
self.delete = self._client.trigger_instances.manage.delete
|
|
self.enable = functools.partial(
|
|
self._client.trigger_instances.manage.update,
|
|
status="enable",
|
|
)
|
|
self.disable = functools.partial(
|
|
self._client.trigger_instances.manage.update,
|
|
status="disable",
|
|
)
|
|
|
|
def set_webhook_subscription(
|
|
self,
|
|
*,
|
|
webhook_url: str,
|
|
enabled_events: t.Optional[t.Sequence[str]] = None,
|
|
version: t.Union[WebhookVersion, str] = WebhookVersion.V3,
|
|
) -> WebhookSubscription:
|
|
"""
|
|
Create or update the project webhook subscription used for webhook delivery.
|
|
|
|
If a subscription already exists, the first subscription is updated. Otherwise a
|
|
new subscription is created. By default this subscribes to V3 trigger message
|
|
events.
|
|
|
|
Example:
|
|
composio.triggers.set_webhook_subscription(
|
|
webhook_url=f"{APP_URL}/webhooks/composio",
|
|
)
|
|
"""
|
|
if not webhook_url:
|
|
raise exceptions.ValidationError("please provide a valid `webhook_url`")
|
|
|
|
events = list(
|
|
DEFAULT_WEBHOOK_SUBSCRIPTION_EVENTS
|
|
if enabled_events is None
|
|
else enabled_events
|
|
)
|
|
if len(events) == 0:
|
|
raise exceptions.ValidationError(
|
|
"please provide at least one enabled event"
|
|
)
|
|
|
|
version_value = (
|
|
version.value if isinstance(version, WebhookVersion) else version
|
|
)
|
|
body = {
|
|
"webhook_url": webhook_url,
|
|
"enabled_events": events,
|
|
"version": version_value,
|
|
}
|
|
|
|
existing = self._client.get(
|
|
WEBHOOK_SUBSCRIPTIONS_PATH,
|
|
cast_to=object,
|
|
options={"params": {"limit": 1}},
|
|
)
|
|
subscription_id = self._first_webhook_subscription_id(existing)
|
|
|
|
if subscription_id:
|
|
return self._normalize_webhook_subscription(
|
|
self._client.patch(
|
|
f"{WEBHOOK_SUBSCRIPTIONS_PATH}/{subscription_id}",
|
|
cast_to=object,
|
|
body=body,
|
|
)
|
|
)
|
|
|
|
return self._normalize_webhook_subscription(
|
|
self._client.post(
|
|
WEBHOOK_SUBSCRIPTIONS_PATH,
|
|
cast_to=object,
|
|
body=body,
|
|
),
|
|
)
|
|
|
|
@staticmethod
|
|
def _normalize_webhook_subscription(raw: object) -> WebhookSubscription:
|
|
"""Build a typed :class:`WebhookSubscription` from the raw API response.
|
|
|
|
Maps explicitly (accepting either snake_case or camelCase wire keys)
|
|
instead of ``cast``-ing the raw object, so the returned dict always
|
|
matches the declared shape and a shift in the wire format surfaces as a
|
|
normalized field rather than a ``KeyError`` at the call site.
|
|
"""
|
|
data = raw if isinstance(raw, dict) else {}
|
|
|
|
def _first_str(*keys: str) -> t.Optional[str]:
|
|
for key in keys:
|
|
value = data.get(key)
|
|
if isinstance(value, str) and value:
|
|
return value
|
|
return None
|
|
|
|
def _str_list(*keys: str) -> t.List[str]:
|
|
for key in keys:
|
|
value = data.get(key)
|
|
if isinstance(value, list):
|
|
return [item for item in value if isinstance(item, str)]
|
|
return []
|
|
|
|
result: WebhookSubscription = {
|
|
"id": _first_str("id") or "",
|
|
"webhook_url": _first_str("webhook_url", "webhookUrl") or "",
|
|
"version": _first_str("version") or WebhookVersion.V3.value,
|
|
"enabled_events": _str_list("enabled_events", "enabledEvents"),
|
|
}
|
|
secret = _first_str("secret")
|
|
if secret is not None:
|
|
result["secret"] = secret
|
|
created_at = _first_str("created_at", "createdAt")
|
|
if created_at is not None:
|
|
result["created_at"] = created_at
|
|
updated_at = _first_str("updated_at", "updatedAt")
|
|
if updated_at is not None:
|
|
result["updated_at"] = updated_at
|
|
return result
|
|
|
|
@staticmethod
|
|
def _first_webhook_subscription_id(response: object) -> t.Optional[str]:
|
|
if not isinstance(response, dict):
|
|
return None
|
|
|
|
items = response.get("items")
|
|
if not isinstance(items, list) or len(items) == 0:
|
|
return None
|
|
|
|
first = items[0]
|
|
if not isinstance(first, dict):
|
|
return None
|
|
|
|
subscription_id = first.get("id")
|
|
return subscription_id if isinstance(subscription_id, str) else None
|
|
|
|
def get_type(self, slug: str) -> TriggersTypeRetrieveResponse:
|
|
"""
|
|
Get a trigger type by its slug
|
|
Uses the global toolkit version provided when initializing composio instance to fetch trigger for specific toolkit version
|
|
|
|
:param slug: The slug of the trigger type
|
|
:return: The trigger type
|
|
"""
|
|
return self._client.triggers_types.retrieve(
|
|
slug=slug, toolkit_versions=none_to_omit(self._toolkit_versions)
|
|
)
|
|
|
|
def list_active(
|
|
self,
|
|
trigger_ids: t.Optional[list[str]] = None,
|
|
trigger_names: t.Optional[list[str]] = None,
|
|
auth_config_ids: t.Optional[list[str]] = None,
|
|
connected_account_ids: t.Optional[list[str]] = None,
|
|
show_disabled: t.Optional[bool] = None,
|
|
limit: t.Optional[int] = None,
|
|
cursor: t.Optional[str] = None,
|
|
):
|
|
"""
|
|
List all active triggers
|
|
|
|
:param trigger_ids: List of trigger IDs to filter by
|
|
:param trigger_names: List of trigger names to filter by
|
|
:param auth_config_ids: List of auth config IDs to filter by
|
|
:param connected_account_ids: List of connected account IDs to filter by
|
|
:param show_disabled: Whether to show disabled triggers
|
|
:param limit: Limit the number of triggers to return
|
|
:param cursor: Cursor for pagination. Use the nextCursor from the response to get the next page.
|
|
:return: List of active triggers
|
|
"""
|
|
return self._client.trigger_instances.list_active(
|
|
query_trigger_ids_1=trigger_ids,
|
|
query_trigger_names_1=trigger_names,
|
|
query_auth_config_ids_1=auth_config_ids,
|
|
query_connected_account_ids_1=connected_account_ids,
|
|
query_show_disabled_1=show_disabled,
|
|
limit=limit if limit is not None else omit,
|
|
cursor=cursor if cursor is not None else omit,
|
|
)
|
|
|
|
def list(
|
|
self,
|
|
*,
|
|
cursor: t.Optional[str] = None,
|
|
limit: t.Optional[int] = None,
|
|
toolkit_slugs: t.Optional[list[str]] = None,
|
|
):
|
|
"""
|
|
List all the trigger types.
|
|
|
|
:param cursor: The cursor for pagination
|
|
:param limit: The maximum number of trigger types to return
|
|
:param toolkit_slugs: Filter by toolkit slugs
|
|
:return: The list of trigger types
|
|
"""
|
|
return self._client.triggers_types.list(
|
|
cursor=none_to_omit(cursor),
|
|
limit=none_to_omit(limit),
|
|
toolkit_slugs=none_to_omit(toolkit_slugs),
|
|
toolkit_versions=none_to_omit(self._toolkit_versions),
|
|
)
|
|
|
|
@t.overload
|
|
def create(
|
|
self,
|
|
slug: str,
|
|
*,
|
|
connected_account_id: str,
|
|
trigger_config: t.Optional[t.Dict[str, t.Any]] = None,
|
|
) -> trigger_instance_upsert_response.TriggerInstanceUpsertResponse: ...
|
|
|
|
@t.overload
|
|
def create(
|
|
self,
|
|
slug: str,
|
|
*,
|
|
user_id: str,
|
|
trigger_config: t.Optional[t.Dict[str, t.Any]] = None,
|
|
) -> trigger_instance_upsert_response.TriggerInstanceUpsertResponse: ...
|
|
|
|
def create(
|
|
self,
|
|
slug: str,
|
|
*,
|
|
user_id: t.Optional[str] = None,
|
|
connected_account_id: t.Optional[str] = None,
|
|
trigger_config: t.Optional[t.Dict[str, t.Any]] = None,
|
|
) -> trigger_instance_upsert_response.TriggerInstanceUpsertResponse:
|
|
"""
|
|
Create a trigger instance
|
|
|
|
:param slug: The slug of the trigger
|
|
:param user_id: The ID of the user that owns the connected account. When
|
|
``connected_account_id`` is omitted, the backend resolves the first
|
|
active connection for this user and the trigger's toolkit.
|
|
:param connected_account_id: The ID of the connected account to pin
|
|
:param trigger_config: The configuration of the trigger
|
|
:return: The trigger instance
|
|
"""
|
|
# Treat blank strings as missing so an empty or whitespace-only
|
|
# `user_id` / `connected_account_id` does not slip past the guard below
|
|
# or reach the backend (parity with the TypeScript SDK's `userId` check).
|
|
if isinstance(user_id, str) and not user_id.strip():
|
|
user_id = None
|
|
if isinstance(connected_account_id, str) and not connected_account_id.strip():
|
|
connected_account_id = None
|
|
|
|
if user_id is None and connected_account_id is None:
|
|
raise exceptions.InvalidParams(
|
|
"please provide valid `connected_account_id` or `user_id`"
|
|
)
|
|
|
|
# Validate the trigger slug up-front so callers get a clear
|
|
# `TriggerTypeNotFound` (parity with the TypeScript SDK).
|
|
try:
|
|
self.get_type(slug=slug)
|
|
except APIStatusError as error:
|
|
if error.status_code in (400, 404):
|
|
raise exceptions.TriggerTypeNotFound(
|
|
f"Trigger type {slug} not found"
|
|
) from error
|
|
raise
|
|
|
|
# Pass user_id straight through: when connected_account_id is omitted the
|
|
# backend resolves the first active connection for this user and the
|
|
# trigger's toolkit (parity with tool execution). When 2FA is enabled and
|
|
# connected_account_id is pinned, the backend validates that user_id owns it.
|
|
return self._client.trigger_instances.upsert(
|
|
slug=slug,
|
|
connected_account_id=none_to_omit(connected_account_id),
|
|
toolkit_versions=self._toolkit_versions,
|
|
body_trigger_config_1=none_to_omit(trigger_config),
|
|
user_id=none_to_omit(user_id),
|
|
)
|
|
|
|
def subscribe(self, timeout: float = 15.0) -> TriggerSubscription:
|
|
"""
|
|
Subscribe to a trigger and receive trigger events.
|
|
|
|
:param timeout: The timeout to wait for the subscription to be established.
|
|
:return: The trigger subscription handler.
|
|
"""
|
|
return _SubcriptionBuilder(client=self._client).connect(timeout=timeout)
|
|
|
|
def verify_webhook(
|
|
self,
|
|
*,
|
|
id: str,
|
|
payload: str,
|
|
secret: str,
|
|
signature: str,
|
|
timestamp: str,
|
|
tolerance: int = 300,
|
|
) -> VerifyWebhookResult:
|
|
"""
|
|
Verify an incoming webhook payload and signature.
|
|
|
|
This method validates that the webhook request is authentic by:
|
|
1. Validating the webhook timestamp is within the tolerance window
|
|
2. Verifying the HMAC-SHA256 signature using the correct algorithm
|
|
3. Parsing the payload and detecting the webhook version (V1, V2, or V3)
|
|
|
|
:param id: The webhook message ID from the 'webhook-id' header (format: 'msg_xxx')
|
|
:param payload: The raw webhook payload as a string (request body)
|
|
:param secret: The webhook secret used to sign the payload (from Composio dashboard)
|
|
:param signature: The signature from the 'webhook-signature' header (format: 'v1,base64EncodedSignature')
|
|
:param timestamp: The webhook timestamp from the 'webhook-timestamp' header (Unix seconds)
|
|
:param tolerance: Maximum allowed age of the webhook in seconds (default: 300 = 5 minutes).
|
|
Set to 0 to disable timestamp validation.
|
|
:return: VerifyWebhookResult containing version, normalized payload, and raw payload
|
|
:raises WebhookSignatureVerificationError: If the signature verification fails
|
|
:raises WebhookPayloadError: If the payload cannot be parsed or is invalid
|
|
|
|
Example:
|
|
# In a Flask webhook handler
|
|
@app.route('/webhook', methods=['POST'])
|
|
def webhook():
|
|
try:
|
|
result = composio.triggers.verify_webhook(
|
|
id=request.headers.get('webhook-id', ''),
|
|
payload=request.get_data(as_text=True),
|
|
signature=request.headers.get('webhook-signature', ''),
|
|
timestamp=request.headers.get('webhook-timestamp', ''),
|
|
secret=os.environ['COMPOSIO_WEBHOOK_SECRET'],
|
|
)
|
|
|
|
# Process the verified payload
|
|
print(f"Version: {result['version']}")
|
|
print(f"Received trigger: {result['payload']['trigger_slug']}")
|
|
return 'OK', 200
|
|
except WebhookSignatureVerificationError:
|
|
return 'Unauthorized', 401
|
|
"""
|
|
# Validate timestamp if tolerance is set
|
|
if tolerance > 0:
|
|
self._validate_webhook_timestamp_header(timestamp, tolerance)
|
|
|
|
# Verify signature using the correct algorithm
|
|
self._verify_webhook_signature(
|
|
webhook_id=id,
|
|
webhook_timestamp=timestamp,
|
|
payload=payload,
|
|
signature=signature,
|
|
secret=secret,
|
|
)
|
|
|
|
# Parse and detect version
|
|
version, raw_payload, normalized_payload = self._parse_webhook_payload(payload)
|
|
|
|
return {
|
|
"version": version,
|
|
"payload": normalized_payload,
|
|
"raw_payload": raw_payload,
|
|
}
|
|
|
|
def parse(
|
|
self,
|
|
request: t.Any = None,
|
|
*,
|
|
body: t.Union[str, bytes, t.Mapping[str, t.Any], None] = None,
|
|
headers: t.Union[t.Mapping[str, t.Any], None] = None,
|
|
verify_secret: t.Union[str, None, Omit] = omit,
|
|
tolerance: int = 300,
|
|
) -> VerifyWebhookResult:
|
|
"""
|
|
Parse an incoming webhook request into a typed, normalized trigger payload.
|
|
|
|
Pass a framework request object, or pass ``body=`` and ``headers=``
|
|
explicitly. When ``verify_secret`` is provided, the SDK verifies the
|
|
webhook signature before returning the normalized trigger payload. When
|
|
it is omitted, the SDK parses the body without verifying the signature.
|
|
|
|
``request`` may be any object exposing the request body and headers, such
|
|
as a Flask, Django, or FastAPI request. The body is read from ``.body``
|
|
(or ``.data`` / ``.get_data()``), and the headers are read from
|
|
``.headers``. Because this SDK is synchronous, async frameworks must pass
|
|
an already-read raw body, for example via ``body=await request.body()``.
|
|
|
|
:param request: The incoming webhook request object (Flask/Django/FastAPI)
|
|
:param body: The raw request body (str/bytes/parsed mapping); overrides ``request``
|
|
:param headers: The request headers as a mapping; overrides ``request``
|
|
:param verify_secret: Webhook secret; when set, the signature is verified.
|
|
Omit it entirely to parse without verification. Passing a present-but-empty
|
|
value (e.g. an unset ``COMPOSIO_WEBHOOK_SECRET``) raises rather than
|
|
silently skipping verification.
|
|
:param tolerance: Max webhook age in seconds (only used when verifying)
|
|
:return: VerifyWebhookResult containing version, normalized payload, and raw payload
|
|
:raises ValidationError: If ``verify_secret`` is empty, or is set but signature headers are missing
|
|
:raises WebhookSignatureVerificationError: If signature verification fails
|
|
:raises WebhookPayloadError: If the payload cannot be parsed
|
|
|
|
Example:
|
|
# Flask: verify the signature
|
|
@app.route('/webhooks/composio', methods=['POST'])
|
|
def webhook():
|
|
try:
|
|
result = composio.triggers.parse(
|
|
request,
|
|
verify_secret=os.environ['COMPOSIO_WEBHOOK_SECRET'],
|
|
)
|
|
print(f"Trigger: {result['payload']['trigger_slug']}")
|
|
print(f"Event data: {result['payload']['payload']}")
|
|
return 'OK', 200
|
|
except exceptions.WebhookSignatureVerificationError:
|
|
return 'Unauthorized', 401
|
|
|
|
# FastAPI: parse without verifying after reading the async body
|
|
@app.post('/webhooks/composio')
|
|
async def webhook(request: Request):
|
|
raw = await request.body()
|
|
result = composio.triggers.parse(body=raw, headers=request.headers)
|
|
return {'trigger': result['payload']['trigger_slug']}
|
|
"""
|
|
raw_body = body if body is not None else self._extract_request_body(request)
|
|
raw_headers = (
|
|
headers if headers is not None else getattr(request, "headers", None)
|
|
)
|
|
|
|
payload = self._body_to_str(raw_body)
|
|
|
|
# Distinguish "caller omitted verify_secret" (explicit opt-out) from
|
|
# "caller passed verify_secret but it resolved to empty" (almost always
|
|
# an unset COMPOSIO_WEBHOOK_SECRET). The latter must fail loudly rather
|
|
# than silently skip verification and accept forged events.
|
|
if isinstance(verify_secret, Omit):
|
|
version, raw_payload, normalized_payload = self._parse_webhook_payload(
|
|
payload
|
|
)
|
|
return {
|
|
"version": version,
|
|
"payload": normalized_payload,
|
|
"raw_payload": raw_payload,
|
|
}
|
|
|
|
if not verify_secret:
|
|
raise exceptions.ValidationError(
|
|
"Cannot verify webhook: `verify_secret` was provided but is empty — "
|
|
"your COMPOSIO_WEBHOOK_SECRET is likely unset. Set the secret, or omit "
|
|
"`verify_secret` entirely to parse without verification."
|
|
)
|
|
|
|
# Secret provided: signature headers are required to verify.
|
|
webhook_id = self._get_header(raw_headers, "webhook-id")
|
|
timestamp = self._get_header(raw_headers, "webhook-timestamp")
|
|
signature = self._get_header(raw_headers, "webhook-signature")
|
|
|
|
missing = [
|
|
name
|
|
for name, value in (
|
|
("webhook-id", webhook_id),
|
|
("webhook-timestamp", timestamp),
|
|
("webhook-signature", signature),
|
|
)
|
|
if not value
|
|
]
|
|
if missing:
|
|
raise exceptions.ValidationError(
|
|
"Cannot verify webhook: missing signature header(s) "
|
|
f"{', '.join(repr(name) for name in missing)}. "
|
|
"Pass the raw, unparsed request body and ensure the Composio "
|
|
"signature headers (webhook-id, webhook-timestamp, "
|
|
"webhook-signature) are forwarded to triggers.parse(). "
|
|
"To parse without verifying, omit `verify_secret`."
|
|
)
|
|
|
|
return self.verify_webhook(
|
|
id=t.cast(str, webhook_id),
|
|
payload=payload,
|
|
secret=verify_secret,
|
|
signature=t.cast(str, signature),
|
|
timestamp=t.cast(str, timestamp),
|
|
tolerance=tolerance,
|
|
)
|
|
|
|
@staticmethod
|
|
def _extract_request_body(request: t.Any) -> t.Union[str, bytes, None]:
|
|
"""Read the raw body from a framework request object.
|
|
|
|
Tries the common attributes used by Flask (``get_data``/``data``),
|
|
Django (``body``) and similar. Returns ``None`` when nothing usable is
|
|
found so the caller surfaces a clear parse error.
|
|
"""
|
|
if request is None:
|
|
return None
|
|
|
|
# Flask: request.get_data() returns the raw body bytes.
|
|
get_data = getattr(request, "get_data", None)
|
|
if callable(get_data):
|
|
return get_data()
|
|
|
|
for attr in ("body", "data"):
|
|
value = getattr(request, attr, None)
|
|
if value is not None and not callable(value):
|
|
return value
|
|
|
|
return None
|
|
|
|
@staticmethod
|
|
def _body_to_str(body: t.Union[str, bytes, t.Mapping[str, t.Any], None]) -> str:
|
|
"""Coerce a body (str, bytes, parsed mapping, or None) into a raw string."""
|
|
if body is None:
|
|
return ""
|
|
if isinstance(body, str):
|
|
return body
|
|
if isinstance(body, (bytes, bytearray)):
|
|
return bytes(body).decode("utf-8")
|
|
# Already-parsed mapping (e.g. a framework that pre-parsed JSON). Note:
|
|
# re-serializing cannot reproduce the exact signed bytes, so signature
|
|
# verification on a pre-parsed body is best-effort only.
|
|
return json.dumps(body)
|
|
|
|
@staticmethod
|
|
def _get_header(headers: t.Any, name: str) -> t.Optional[str]:
|
|
"""Read a header value case-insensitively from a headers mapping.
|
|
|
|
Supports both plain ``dict`` headers and framework header objects
|
|
(Werkzeug/Django ``HttpHeaders``) that expose a ``get`` method, which is
|
|
already case-insensitive.
|
|
"""
|
|
if headers is None:
|
|
return None
|
|
|
|
get = getattr(headers, "get", None)
|
|
if callable(get):
|
|
value = get(name)
|
|
if value is not None:
|
|
return str(value)
|
|
|
|
# Fall back to a manual case-insensitive scan for plain mappings.
|
|
try:
|
|
items = headers.items()
|
|
except AttributeError:
|
|
return None
|
|
|
|
target = name.lower()
|
|
for key, value in items:
|
|
if isinstance(key, str) and key.lower() == target:
|
|
return str(value) if value is not None else None
|
|
return None
|
|
|
|
def _verify_webhook_signature(
|
|
self,
|
|
*,
|
|
webhook_id: str,
|
|
webhook_timestamp: str,
|
|
payload: str,
|
|
signature: str,
|
|
secret: str,
|
|
) -> None:
|
|
"""
|
|
Verify the HMAC-SHA256 signature of a webhook payload.
|
|
|
|
The signature is computed as: HMAC-SHA256(webhookId.webhookTimestamp.payload, secret)
|
|
and then base64 encoded with a 'v1,' prefix.
|
|
|
|
:param webhook_id: The webhook message ID from header
|
|
:param webhook_timestamp: The webhook timestamp from header
|
|
:param payload: The raw webhook payload
|
|
:param signature: The signature to verify (format: 'v1,base64EncodedSignature')
|
|
:param secret: The webhook secret
|
|
:raises WebhookSignatureVerificationError: If verification fails
|
|
"""
|
|
if not payload:
|
|
raise exceptions.WebhookSignatureVerificationError(
|
|
"No webhook payload was provided."
|
|
)
|
|
|
|
if not signature:
|
|
raise exceptions.WebhookSignatureVerificationError(
|
|
"No signature header value was provided. "
|
|
"Please pass the value of the webhook signature header."
|
|
)
|
|
|
|
if not secret:
|
|
raise exceptions.WebhookSignatureVerificationError(
|
|
"No webhook secret was provided. "
|
|
"You can find your webhook secret in your Composio dashboard."
|
|
)
|
|
|
|
if not webhook_id:
|
|
raise exceptions.WebhookSignatureVerificationError(
|
|
"No webhook ID was provided. "
|
|
"Please pass the value of the 'webhook-id' header."
|
|
)
|
|
|
|
if not webhook_timestamp:
|
|
raise exceptions.WebhookSignatureVerificationError(
|
|
"No webhook timestamp was provided. "
|
|
"Please pass the value of the 'webhook-timestamp' header."
|
|
)
|
|
|
|
# Parse signature header - format is "v1,base64Sig" or "v1,sig1 v1,sig2"
|
|
# Split by space to handle multiple signatures
|
|
signature_parts = signature.split(" ")
|
|
v1_signatures: t.List[str] = []
|
|
|
|
for part in signature_parts:
|
|
if part.startswith("v1,"):
|
|
v1_signatures.append(part[3:]) # Remove "v1," prefix
|
|
|
|
if not v1_signatures:
|
|
raise exceptions.WebhookSignatureVerificationError(
|
|
"No valid v1 signature found in the signature header. "
|
|
"Expected format: 'v1,base64EncodedSignature'"
|
|
)
|
|
|
|
# Construct the string to sign: webhookId.webhookTimestamp.payload
|
|
to_sign = f"{webhook_id}.{webhook_timestamp}.{payload}"
|
|
|
|
# Compute expected signature
|
|
expected_signature_bytes = hmac.new(
|
|
key=secret.encode("utf-8"),
|
|
msg=to_sign.encode("utf-8"),
|
|
digestmod=hashlib.sha256,
|
|
).digest()
|
|
expected_signature_b64 = base64.b64encode(expected_signature_bytes).decode(
|
|
"utf-8"
|
|
)
|
|
|
|
# Check if any of the provided signatures match (timing-safe)
|
|
for provided_sig in v1_signatures:
|
|
try:
|
|
if hmac.compare_digest(provided_sig, expected_signature_b64):
|
|
return # Signature is valid
|
|
except Exception:
|
|
continue # Invalid signature format, try next
|
|
|
|
raise exceptions.WebhookSignatureVerificationError(
|
|
"The signature provided is invalid."
|
|
)
|
|
|
|
def _validate_webhook_timestamp_header(
|
|
self, timestamp: str, tolerance: int
|
|
) -> None:
|
|
"""
|
|
Validate that the webhook timestamp header is within the allowed tolerance.
|
|
|
|
:param timestamp: The webhook timestamp from header (Unix seconds)
|
|
:param tolerance: Maximum allowed age in seconds
|
|
:raises WebhookSignatureVerificationError: If timestamp is outside tolerance
|
|
:raises WebhookPayloadError: If timestamp format is invalid
|
|
"""
|
|
try:
|
|
timestamp_seconds = int(timestamp)
|
|
except (ValueError, TypeError) as e:
|
|
raise exceptions.WebhookPayloadError(
|
|
f"Invalid webhook timestamp: {timestamp}. "
|
|
"Expected Unix timestamp in seconds."
|
|
) from e
|
|
|
|
current_time = int(time.time())
|
|
time_difference = abs(current_time - timestamp_seconds)
|
|
|
|
if time_difference > tolerance:
|
|
raise exceptions.WebhookSignatureVerificationError(
|
|
f"The webhook timestamp is outside the allowed tolerance. "
|
|
f"The webhook was sent {time_difference} seconds ago, "
|
|
f"but the maximum allowed age is {tolerance} seconds."
|
|
)
|
|
|
|
def _parse_webhook_payload(
|
|
self, payload: str
|
|
) -> t.Tuple[WebhookVersion, WebhookPayload, TriggerEvent]:
|
|
"""
|
|
Parse webhook payload and detect version.
|
|
|
|
:param payload: The raw webhook payload string
|
|
:return: Tuple of (version, raw_payload, normalized_payload)
|
|
:raises WebhookPayloadError: If payload cannot be parsed
|
|
"""
|
|
try:
|
|
data = json.loads(payload)
|
|
except json.JSONDecodeError as e:
|
|
raise exceptions.WebhookPayloadError(
|
|
f"Failed to parse webhook payload as JSON: {e}"
|
|
) from e
|
|
|
|
# Try V3 first — same envelope the realtime channel uses, so reuse the
|
|
# shared detector/normalizer to keep the two paths from drifting.
|
|
if _is_v3_envelope(data):
|
|
return (
|
|
WebhookVersion.V3,
|
|
t.cast(WebhookPayloadV3, data),
|
|
_build_trigger_event_from_v3(t.cast(WebhookPayloadV3, data)),
|
|
)
|
|
|
|
# Try V2 (has 'type', 'timestamp', 'data' with nested fields)
|
|
if (
|
|
isinstance(data, dict)
|
|
and "type" in data
|
|
and "timestamp" in data
|
|
and "data" in data
|
|
and isinstance(data.get("data"), dict)
|
|
and "connection_id" in data.get("data", {})
|
|
):
|
|
return (
|
|
WebhookVersion.V2,
|
|
t.cast(WebhookPayloadV2, data),
|
|
self._normalize_v2_payload(t.cast(WebhookPayloadV2, data)),
|
|
)
|
|
|
|
# Try V1 (has 'trigger_name', 'connection_id', 'trigger_id', 'payload')
|
|
if (
|
|
isinstance(data, dict)
|
|
and "trigger_name" in data
|
|
and "connection_id" in data
|
|
and "trigger_id" in data
|
|
and "payload" in data
|
|
):
|
|
return (
|
|
WebhookVersion.V1,
|
|
t.cast(WebhookPayloadV1, data),
|
|
self._normalize_v1_payload(t.cast(WebhookPayloadV1, data)),
|
|
)
|
|
|
|
raise exceptions.WebhookPayloadError(
|
|
"Webhook payload does not match any known version (V1, V2, or V3). "
|
|
"Please ensure the payload structure is correct."
|
|
)
|
|
|
|
def _normalize_v1_payload(self, data: WebhookPayloadV1) -> TriggerEvent:
|
|
"""Normalize V1 payload to TriggerEvent format."""
|
|
return t.cast(
|
|
TriggerEvent,
|
|
{
|
|
"id": data["trigger_id"],
|
|
"uuid": data["trigger_id"],
|
|
"user_id": "", # V1 doesn't have user_id
|
|
"toolkit_slug": "", # V1 doesn't have toolkit_slug
|
|
"trigger_slug": data["trigger_name"],
|
|
"metadata": {
|
|
"id": data["trigger_id"],
|
|
"uuid": data["trigger_id"],
|
|
"toolkit_slug": "",
|
|
"trigger_slug": data["trigger_name"],
|
|
"trigger_data": None,
|
|
"trigger_config": {},
|
|
"connected_account": {
|
|
"id": data["connection_id"],
|
|
"uuid": data["connection_id"],
|
|
"auth_config_id": "",
|
|
"auth_config_uuid": "",
|
|
"user_id": "",
|
|
"status": "ACTIVE",
|
|
},
|
|
},
|
|
"payload": data["payload"],
|
|
"original_payload": None,
|
|
},
|
|
)
|
|
|
|
def _normalize_v2_payload(self, data: WebhookPayloadV2) -> TriggerEvent:
|
|
"""Normalize V2 payload to TriggerEvent format."""
|
|
payload_data = data["data"]
|
|
return t.cast(
|
|
TriggerEvent,
|
|
{
|
|
"id": payload_data.get(
|
|
"trigger_nano_id", payload_data.get("trigger_id", "")
|
|
),
|
|
"uuid": payload_data.get("trigger_id", ""),
|
|
"user_id": payload_data.get("user_id", ""),
|
|
"toolkit_slug": data["type"].upper() if data.get("type") else "",
|
|
"trigger_slug": data["type"].upper() if data.get("type") else "",
|
|
"metadata": {
|
|
"id": payload_data.get("trigger_nano_id", ""),
|
|
"uuid": payload_data.get("trigger_id", ""),
|
|
"toolkit_slug": data["type"].upper() if data.get("type") else "",
|
|
"trigger_slug": data["type"].upper() if data.get("type") else "",
|
|
"trigger_data": None,
|
|
"trigger_config": {},
|
|
"connected_account": {
|
|
"id": payload_data.get("connection_nano_id", ""),
|
|
"uuid": payload_data.get("connection_id", ""),
|
|
"auth_config_id": "",
|
|
"auth_config_uuid": "",
|
|
"user_id": payload_data.get("user_id", ""),
|
|
"status": "ACTIVE",
|
|
},
|
|
},
|
|
"payload": {
|
|
k: v
|
|
for k, v in payload_data.items()
|
|
if k
|
|
not in (
|
|
"connection_id",
|
|
"connection_nano_id",
|
|
"trigger_nano_id",
|
|
"trigger_id",
|
|
"user_id",
|
|
)
|
|
},
|
|
"original_payload": None,
|
|
},
|
|
)
|