vodorod/Dorod-Sky
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Webhook Replay Test URL (#3769)

Browse Source 
Co-authored-by: Shuchang Zheng <wintonzheng0325@gmail.com>
This commit is contained in:
Marc Kelechava
2025-10-22 14:26:14 -07:00
committed by GitHub
parent e8472df6d1
commit aeefc301ed
8 changed files with 964 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
skyvern/schemas/webhooks.py
View File

@@ -15,3 +15,31 @@ class TestWebhookResponse(BaseModel):
response_body: str = Field(..., description="First 2KB of the response body")
headers_sent: dict[str, str] = Field(..., description="Headers sent with the webhook request")
error: str | None = Field(None, description="Error message if the request failed")
class RunWebhookPreviewResponse(BaseModel):
run_id: str = Field(..., description="Identifier of the run whose payload is being replayed")
run_type: str = Field(..., description="Run type associated with the payload")
default_webhook_url: str | None = Field(None, description="Webhook URL stored on the original run configuration")
payload: str = Field(..., description="JSON payload that was delivered when the run completed")
headers: dict[str, str] = Field(..., description="Signed headers that would accompany the replayed webhook")
class RunWebhookReplayRequest(BaseModel):
override_webhook_url: str | None = Field(
None,
description="Optional webhook URL to send the payload to instead of the stored configuration",
)
class RunWebhookReplayResponse(BaseModel):
run_id: str = Field(..., description="Identifier of the run that was replayed")
run_type: str = Field(..., description="Run type associated with the payload")
default_webhook_url: str | None = Field(None, description="Webhook URL stored on the original run configuration")
target_webhook_url: str | None = Field(None, description="Webhook URL that the replay attempted to reach")
payload: str = Field(..., description="JSON payload that was delivered during the replay attempt")
headers: dict[str, str] = Field(..., description="Signed headers that were generated for the replay attempt")
status_code: int | None = Field(None, description="HTTP status code returned by the webhook receiver, if available")
latency_ms: int | None = Field(None, description="Round-trip latency in milliseconds for the replay attempt")
response_body: str | None = Field(None, description="Body returned by the webhook receiver (truncated to 2KB)")
error: str | None = Field(None, description="Error message if the replay attempt failed")