apunkt/dograh
1
0
Fork 0
mirror of https://github.com/dograh-hq/dograh.git synced 2026-07-13 11:22:14 +02:00
Code Issues Projects Releases Packages Wiki Activity

fix: scope workflow run to org rather than user

Browse source
This commit is contained in:
Abhishek Kumar 2026-07-10 16:52:43 +05:30
parent 4989bab1e9
commit 43737c67dc
29 changed files with 218 additions and 158 deletions
Download patch file Download diff file Expand all files Collapse all files

1
api/routes/agent_stream.py
View file

@ -117,7 +117,6 @@ async def agent_stream_websocket(
websocket,
organization_id=workflow.organization_id,
workflow_id=workflow.id,
user_id=workflow.user_id,
workflow_run_id=workflow_run.id,
params=params,
)

7
api/routes/public_agent.py
View file

@ -303,13 +303,12 @@ async def _execute_resolved_target(
webhook_url = (
f"{backend_endpoint}/api/v1/telephony/{webhook_endpoint}"
f"?workflow_id={target.workflow.id}"
f"&user_id={execution_user_id}"
f"&workflow_run_id={workflow_run.id}"
f"&organization_id={target.organization_id}"
)
# 10. Initiate call via telephony provider. workflow_id and user_id are
# required by providers that build the media WebSocket URL at dial time
# 10. Initiate call via telephony provider. workflow_id and organization_id
# are required by providers that build the media WebSocket URL at dial time
# (e.g. Telnyx, Cloudonix); without them the URL contains "None/None" and
# the stream connection fails.
try:
@ -318,7 +317,7 @@ async def _execute_resolved_target(
webhook_url=webhook_url,
workflow_run_id=workflow_run.id,
workflow_id=target.workflow.id,
user_id=execution_user_id,
organization_id=target.organization_id,
)
except Exception as e:
logger.warning(

67
api/routes/telephony.py
View file

@ -240,12 +240,14 @@ async def initiate_call(
webhook_url = (
f"{backend_endpoint}/api/v1/telephony/{webhook_endpoint}"
f"?workflow_id={workflow.id}"
f"&user_id={execution_user_id}"
f"&workflow_run_id={workflow_run_id}"
f"&organization_id={user.selected_organization_id}"
)
keywords = {"workflow_id": workflow.id, "user_id": execution_user_id}
keywords = {
"workflow_id": workflow.id,
"organization_id": user.selected_organization_id,
}
# Initiate call via provider
result = await provider.initiate_call(
@ -543,13 +545,13 @@ async def websocket_ari_endpoint(websocket: WebSocket):
query params (appended by the v() dial string option in externalMedia).
"""
workflow_id = websocket.query_params.get("workflow_id")
user_id = websocket.query_params.get("user_id")
organization_id = websocket.query_params.get("organization_id")
workflow_run_id = websocket.query_params.get("workflow_run_id")
if not workflow_id or not user_id or not workflow_run_id:
if not workflow_id or not organization_id or not workflow_run_id:
logger.error(
f"ARI WebSocket missing query params: "
f"workflow_id={workflow_id}, user_id={user_id}, workflow_run_id={workflow_run_id}"
f"ARI WebSocket missing query params: workflow_id={workflow_id}, "
f"organization_id={organization_id}, workflow_run_id={workflow_run_id}"
)
await websocket.close(code=4400, reason="Missing required query params")
return
@ -559,39 +561,55 @@ async def websocket_ari_endpoint(websocket: WebSocket):
await websocket.accept(subprotocol="media")
await _handle_telephony_websocket(
websocket, int(workflow_id), int(user_id), int(workflow_run_id)
websocket, int(workflow_id), int(organization_id), int(workflow_run_id)
)
@router.websocket("/ws/{workflow_id}/{user_id}/{workflow_run_id}")
@router.websocket("/ws/{workflow_id}/{organization_id}/{workflow_run_id}")
async def websocket_endpoint(
websocket: WebSocket, workflow_id: int, user_id: int, workflow_run_id: int
websocket: WebSocket, workflow_id: int, organization_id: int, workflow_run_id: int
):
"""WebSocket endpoint for real-time call handling - routes to provider-specific handlers."""
await websocket.accept()
await _handle_telephony_websocket(websocket, workflow_id, user_id, workflow_run_id)
await _handle_telephony_websocket(
websocket, workflow_id, organization_id, workflow_run_id
)
async def _handle_telephony_websocket(
websocket: WebSocket, workflow_id: int, user_id: int, workflow_run_id: int
websocket: WebSocket, workflow_id: int, organization_id: int, workflow_run_id: int
):
"""Shared WebSocket handler logic (connection already accepted)."""
"""Shared WebSocket handler logic (connection already accepted).
TODO(security): ``organization_id`` arrives in the URL the provider dials
back, so it is caller-supplied and unauthenticated — this socket has no
signature check, and the id triple is a guessable bearer capability.
Scoping the lookups below by it prevents an accidental cross-org mismatch,
not a deliberate one. The real fix is a one-shot capability token minted at
run creation and redeemed here through an atomic
``initialized -> running`` compare-and-swap, which would also close the
read-then-write race on the state check further down.
"""
try:
# Set the run context
set_current_run_id(workflow_run_id)
# Get workflow run to determine provider type
workflow_run = await db_client.get_workflow_run(workflow_run_id)
workflow_run = await db_client.get_workflow_run(
workflow_run_id, organization_id=organization_id
)
if not workflow_run:
logger.error(f"Workflow run {workflow_run_id} not found")
logger.error(
f"Workflow run {workflow_run_id} not found for org {organization_id}"
)
await websocket.close(code=4404, reason="Workflow run not found")
return
# Get workflow for organization info. System lookup keyed only on the
# workflow_id (org is derived below) — use the explicit unscoped variant.
workflow = await db_client.get_workflow_by_id(workflow_id)
workflow = await db_client.get_workflow(
workflow_id, organization_id=organization_id
)
if not workflow:
logger.error(f"Workflow {workflow_id} not found")
logger.error(f"Workflow {workflow_id} not found for org {organization_id}")
await websocket.close(code=4404, reason="Workflow not found")
return
if workflow_run.workflow_id != workflow.id:
@ -601,13 +619,6 @@ async def _handle_telephony_websocket(
)
await websocket.close(code=4400, reason="workflow_run_workflow_mismatch")
return
if workflow.user_id != user_id:
logger.error(
f"Telephony websocket user mismatch for workflow {workflow.id}: "
f"got {user_id}, expected {workflow.user_id}"
)
await websocket.close(code=4400, reason="workflow_user_mismatch")
return
# Check workflow run state - only allow 'initialized' state
if workflow_run.state != WorkflowRunState.INITIALIZED.value:
@ -689,7 +700,7 @@ async def _handle_telephony_websocket(
# Delegate to provider-specific handler
await provider.handle_websocket(
websocket, workflow_id, user_id, workflow_run_id
websocket, workflow_id, organization_id, workflow_run_id
)
except WebSocketDisconnect as e:
@ -854,7 +865,7 @@ async def handle_inbound_run(request: Request):
backend_endpoint, wss_backend_endpoint = await get_backend_endpoints()
websocket_url = (
f"{wss_backend_endpoint}/api/v1/telephony/ws/"
f"{workflow_id}/{user_id}/{workflow_run_id}"
f"{workflow_id}/{config.organization_id}/{workflow_run_id}"
)
return await provider_instance.start_inbound_stream(
@ -1022,7 +1033,7 @@ async def handle_inbound_telephony(
backend_endpoint, wss_backend_endpoint = await get_backend_endpoints()
websocket_url = (
f"{wss_backend_endpoint}/api/v1/telephony/ws/"
f"{workflow_id}/{workflow_context['user_id']}/{workflow_run_id}"
f"{workflow_id}/{organization_id}/{workflow_run_id}"
)
response = await provider_instance.start_inbound_stream(

3
api/services/campaign/campaign_call_dispatcher.py
View file

@ -354,7 +354,6 @@ class CampaignCallDispatcher:
webhook_url = (
f"{backend_endpoint}/api/v1/telephony/{webhook_endpoint}"
f"?workflow_id={campaign.workflow_id}"
f"&user_id={campaign.created_by}"
f"&workflow_run_id={workflow_run.id}"
f"&organization_id={campaign.organization_id}"
)
@ -365,7 +364,7 @@ class CampaignCallDispatcher:
workflow_run_id=workflow_run.id,
from_number=from_number,
workflow_id=campaign.workflow_id,
user_id=campaign.created_by,
organization_id=campaign.organization_id,
)
# Store provider type and metadata in gathered_context

42
api/services/pipecat/run_pipeline.py
View file

@ -254,7 +254,7 @@ async def run_pipeline_telephony(
provider_name: str,
workflow_id: int,
workflow_run_id: int,
user_id: int,
organization_id: int,
call_id: str,
transport_kwargs: dict,
) -> None:
@ -268,7 +268,7 @@ async def run_pipeline_telephony(
provider_name=provider_name,
workflow_id=workflow_id,
workflow_run_id=workflow_run_id,
user_id=user_id,
organization_id=organization_id,
call_id=call_id,
transport_kwargs=transport_kwargs,
)
@ -285,7 +285,7 @@ async def _run_pipeline_telephony_impl(
provider_name: str,
workflow_id: int,
workflow_run_id: int,
user_id: int,
organization_id: int,
call_id: str,
transport_kwargs: dict,
) -> None:
@ -300,7 +300,9 @@ async def _run_pipeline_telephony_impl(
provider_name: Stable identifier of the provider (registry key).
workflow_id: Workflow being executed.
workflow_run_id: Workflow run row.
user_id: Owner of the workflow.
organization_id: Tenant owning the workflow and the run. Every lookup
below is scoped by it; the workflow owner is read off the workflow
row and used only for attribution.
call_id: Provider call identifier.
transport_kwargs: Provider-specific kwargs forwarded to the transport
factory (e.g. stream_sid + call_sid for Twilio).
@ -308,12 +310,15 @@ async def _run_pipeline_telephony_impl(
logger.debug(f"Running {provider_name} pipeline for workflow_run {workflow_run_id}")
set_current_run_id(workflow_run_id)
workflow = await db_client.get_workflow(workflow_id, user_id)
if workflow:
set_current_org_id(workflow.organization_id)
workflow = await db_client.get_workflow(
workflow_id, organization_id=organization_id
)
if not workflow:
raise HTTPException(status_code=404, detail="Workflow not found")
set_current_org_id(workflow.organization_id)
ambient_noise_config = None
if workflow and workflow.workflow_configurations:
if workflow.workflow_configurations:
ambient_noise_config = workflow.workflow_configurations.get(
"ambient_noise_configuration"
)
@ -322,9 +327,16 @@ async def _run_pipeline_telephony_impl(
# (test call, campaign dispatch, inbound). Transports use it to load creds
# from the right config row. Falls back to None for legacy runs (transports
# then resolve the org's default config).
workflow_run = await db_client.get_workflow_run(workflow_run_id)
workflow_run = await db_client.get_workflow_run(
workflow_run_id, organization_id=organization_id
)
if not workflow_run:
raise HTTPException(status_code=404, detail="Workflow run not found")
if workflow_run.workflow_id != workflow_id:
raise HTTPException(status_code=400, detail="workflow_run_workflow_mismatch")
telephony_configuration_id = None
if workflow_run and workflow_run.initial_context:
if workflow_run.initial_context:
telephony_configuration_id = workflow_run.initial_context.get(
"telephony_configuration_id"
)
@ -336,11 +348,9 @@ async def _run_pipeline_telephony_impl(
get_effective_ai_model_configuration_for_workflow,
)
run_configs = (
(workflow_run.definition.workflow_configurations or {}) if workflow_run else {}
)
run_configs = workflow_run.definition.workflow_configurations or {}
user_config = await get_effective_ai_model_configuration_for_workflow(
organization_id=workflow.organization_id if workflow else None,
organization_id=workflow.organization_id,
workflow_configurations=run_configs,
)
is_realtime = bool(user_config.is_realtime and user_config.realtime is not None)
@ -364,10 +374,12 @@ async def _run_pipeline_telephony_impl(
transport,
workflow_id,
workflow_run_id,
user_id,
# Attribution only — scoping is driven by organization_id below.
workflow.user_id,
audio_config=audio_config,
workflow_run=workflow_run,
resolved_user_config=user_config,
organization_id=organization_id,
)
except Exception as e:
logger.error(

2
api/services/telephony/README.md
View file

@ -77,7 +77,7 @@ class TelephonyProvider(ABC):
async def verify_webhook_signature(self, url: str, params: Dict[str, Any], signature: str) -> bool
@abstractmethod
async def get_webhook_response(self, workflow_id: int, user_id: int, workflow_run_id: int) -> str
async def get_webhook_response(self, workflow_id: int, organization_id: int, workflow_run_id: int) -> str
```
## Configuration Loading

15
api/services/telephony/ari_manager.py
View file

@ -353,19 +353,18 @@ class ARIConnection:
workflow_run_id = args_dict.get("workflow_run_id")
workflow_id = args_dict.get("workflow_id")
user_id = args_dict.get("user_id")
if not workflow_run_id or not workflow_id or not user_id:
if not workflow_run_id or not workflow_id:
logger.warning(
f"[ARI org={self.organization_id}] StasisStart missing required args: "
f"workflow_run_id={workflow_run_id}, workflow_id={workflow_id}, user_id={user_id}"
f"workflow_run_id={workflow_run_id}, workflow_id={workflow_id}"
)
return
# Start pipeline connection in background task
asyncio.create_task(
self._handle_stasis_start(
channel_id, channel_state, workflow_run_id, workflow_id, user_id
channel_id, channel_state, workflow_run_id, workflow_id
)
)
@ -448,7 +447,6 @@ class ARIConnection:
async def _create_external_media(
self,
workflow_id: str,
user_id: str,
workflow_run_id: str,
channel_id: Optional[str] = None,
) -> str:
@ -464,10 +462,10 @@ class ARIConnection:
the POST and avoid racing against the StasisStart event.
"""
# v() appends URI query params to the websocket_client.conf URL
# e.g. wss://api.dograh.com/ws/ari?workflow_id=1&user_id=2&workflow_run_id=3
# e.g. wss://api.dograh.com/ws/ari?workflow_id=1&organization_id=2&workflow_run_id=3
transport_data = (
f"v(workflow_id={workflow_id},"
f"user_id={user_id},"
f"organization_id={self.organization_id},"
f"workflow_run_id={workflow_run_id})"
)
@ -646,7 +644,6 @@ class ARIConnection:
channel_state,
str(workflow_run.id),
str(inbound_workflow_id),
str(user_id),
)
except Exception as e:
if workflow_run:
@ -668,7 +665,6 @@ class ARIConnection:
channel_state: str,
workflow_run_id: str,
workflow_id: str,
user_id: str,
):
"""Set up external media for a caller channel that has entered Stasis.
@ -712,7 +708,6 @@ class ARIConnection:
# 3. Create the ext media channel with the id we just registered.
created_id = await self._create_external_media(
workflow_id,
user_id,
workflow_run_id,
channel_id=ext_channel_id,
)

14
api/services/telephony/base.py
View file

@ -150,14 +150,15 @@ class TelephonyProvider(ABC):
@abstractmethod
async def get_webhook_response(
self, workflow_id: int, user_id: int, workflow_run_id: int
self, workflow_id: int, organization_id: int, workflow_run_id: int
) -> str:
"""
Generate the initial webhook response for starting a call session.
Args:
workflow_id: The workflow ID
user_id: The user ID
organization_id: The organization owning the workflow; providers
embed it in the media websocket URL they hand back
workflow_run_id: The workflow run ID
Returns:
@ -223,7 +224,7 @@ class TelephonyProvider(ABC):
self,
websocket: "WebSocket",
workflow_id: int,
user_id: int,
organization_id: int,
workflow_run_id: int,
) -> None:
"""
@ -232,10 +233,14 @@ class TelephonyProvider(ABC):
This method encapsulates all provider-specific WebSocket handshake and
message routing logic, keeping the main websocket endpoint clean.
``organization_id`` is the tenant that every workflow/run lookup must be
scoped by. The workflow owner is deliberately not passed in — derive it
from the workflow row where it's needed for attribution.
Args:
websocket: The WebSocket connection
workflow_id: The workflow ID
user_id: The user ID
organization_id: The organization owning the workflow and run
workflow_run_id: The workflow run ID
"""
pass
@ -355,7 +360,6 @@ class TelephonyProvider(ABC):
*,
organization_id: int,
workflow_id: int,
user_id: int,
workflow_run_id: int,
params: Dict[str, str],
) -> None:

11
api/services/telephony/providers/ari/provider.py
View file

@ -99,7 +99,6 @@ class ARIProvider(TelephonyProvider):
[
f"workflow_run_id={workflow_run_id}",
f"workflow_id={kwargs.get('workflow_id', '')}",
f"user_id={kwargs.get('user_id', '')}",
],
)
),
@ -179,7 +178,7 @@ class ARIProvider(TelephonyProvider):
return True
async def get_webhook_response(
self, workflow_id: int, user_id: int, workflow_run_id: int
self, workflow_id: int, organization_id: int, workflow_run_id: int
) -> str:
"""ARI does not use webhook responses - call control is via REST API."""
logger.warning(
@ -241,7 +240,7 @@ class ARIProvider(TelephonyProvider):
self,
websocket: "WebSocket",
workflow_id: int,
user_id: int,
organization_id: int,
workflow_run_id: int,
) -> None:
"""
@ -253,7 +252,9 @@ class ARIProvider(TelephonyProvider):
from api.services.pipecat.run_pipeline import run_pipeline_telephony
# Get channel_id from workflow run context
workflow_run = await db_client.get_workflow_run(workflow_run_id, user_id)
workflow_run = await db_client.get_workflow_run(
workflow_run_id, organization_id=organization_id
)
channel_id = ""
if workflow_run and workflow_run.gathered_context:
channel_id = workflow_run.gathered_context.get("call_id", "")
@ -267,7 +268,7 @@ class ARIProvider(TelephonyProvider):
provider_name=self.PROVIDER_NAME,
workflow_id=workflow_id,
workflow_run_id=workflow_run_id,
user_id=user_id,
organization_id=organization_id,
call_id=channel_id,
transport_kwargs={"channel_id": channel_id},
)

16
api/services/telephony/providers/cloudonix/provider.py
View file

@ -120,7 +120,10 @@ class CloudonixProvider(TelephonyProvider):
logger.info(
f"Selected phone number {from_number} for outbound call to {to_number}"
)
workflow_id, user_id = kwargs["workflow_id"], kwargs["user_id"]
workflow_id, organization_id = (
kwargs["workflow_id"],
kwargs["organization_id"],
)
# Prepare call data using Cloudonix callObject schema
# Note: 'caller-id' is REQUIRED by Cloudonix API
@ -130,7 +133,7 @@ class CloudonixProvider(TelephonyProvider):
"cxml": f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Connect>
<Stream url="{wss_backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{user_id}/{workflow_run_id}"></Stream>
<Stream url="{wss_backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{organization_id}/{workflow_run_id}"></Stream>
</Connect>
<Pause length="40"/>
</Response>""",
@ -408,7 +411,7 @@ class CloudonixProvider(TelephonyProvider):
}
async def get_webhook_response(
self, workflow_id: int, user_id: int, workflow_run_id: int
self, workflow_id: int, organization_id: int, workflow_run_id: int
) -> str:
"""
Dummy implementation - Cloudonix doesn't use webhook responses.
@ -430,7 +433,7 @@ class CloudonixProvider(TelephonyProvider):
self,
websocket: "WebSocket",
workflow_id: int,
user_id: int,
organization_id: int,
workflow_run_id: int,
) -> None:
"""
@ -502,7 +505,7 @@ class CloudonixProvider(TelephonyProvider):
provider_name=self.PROVIDER_NAME,
workflow_id=workflow_id,
workflow_run_id=workflow_run_id,
user_id=user_id,
organization_id=organization_id,
call_id=call_id,
transport_kwargs={"call_id": call_id, "stream_sid": stream_sid},
)
@ -517,7 +520,6 @@ class CloudonixProvider(TelephonyProvider):
*,
organization_id: int,
workflow_id: int,
user_id: int,
workflow_run_id: int,
params: Dict[str, str],
) -> None:
@ -671,7 +673,7 @@ class CloudonixProvider(TelephonyProvider):
provider_name=self.PROVIDER_NAME,
workflow_id=workflow_id,
workflow_run_id=workflow_run_id,
user_id=user_id,
organization_id=organization_id,
call_id=call_session,
transport_kwargs={
"call_id": call_session,

8
api/services/telephony/providers/plivo/provider.py
View file

@ -238,13 +238,13 @@ class PlivoProvider(TelephonyProvider):
return any(hmac.compare_digest(computed, candidate) for candidate in candidates)
async def get_webhook_response(
self, workflow_id: int, user_id: int, workflow_run_id: int
self, workflow_id: int, organization_id: int, workflow_run_id: int
) -> str:
_, wss_backend_endpoint = await get_backend_endpoints()
return f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Stream bidirectional="true" keepCallAlive="true" contentType="audio/x-mulaw;rate=8000">{wss_backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{user_id}/{workflow_run_id}</Stream>
<Stream bidirectional="true" keepCallAlive="true" contentType="audio/x-mulaw;rate=8000">{wss_backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{organization_id}/{workflow_run_id}</Stream>
</Response>"""
async def get_call_cost(self, call_id: str) -> Dict[str, Any]:
@ -309,7 +309,7 @@ class PlivoProvider(TelephonyProvider):
self,
websocket: "WebSocket",
workflow_id: int,
user_id: int,
organization_id: int,
workflow_run_id: int,
) -> None:
from api.services.pipecat.run_pipeline import run_pipeline_telephony
@ -348,7 +348,7 @@ class PlivoProvider(TelephonyProvider):
provider_name=self.PROVIDER_NAME,
workflow_id=workflow_id,
workflow_run_id=workflow_run_id,
user_id=user_id,
organization_id=organization_id,
call_id=call_id,
transport_kwargs={"stream_id": stream_id, "call_id": call_id},
)

3
api/services/telephony/providers/plivo/routes.py
View file

@ -74,7 +74,6 @@ async def _handle_plivo_status_callback(
@router.post("/plivo-xml", include_in_schema=False)
async def handle_plivo_xml_webhook(
workflow_id: int,
user_id: int,
workflow_run_id: int,
organization_id: int,
request: Request,
@ -110,7 +109,7 @@ async def handle_plivo_xml_webhook(
)
response_content = await provider.get_webhook_response(
workflow_id, user_id, workflow_run_id
workflow_id, organization_id, workflow_run_id
)
return HTMLResponse(content=response_content, media_type="application/xml")

10
api/services/telephony/providers/telnyx/provider.py
View file

@ -101,10 +101,10 @@ class TelnyxProvider(TelephonyProvider):
# Build the WebSocket stream URL for inline audio streaming
workflow_id = kwargs.get("workflow_id")
user_id = kwargs.get("user_id")
organization_id = kwargs.get("organization_id")
stream_url = (
f"{wss_backend_endpoint}/api/v1/telephony/ws"
f"/{workflow_id}/{user_id}/{workflow_run_id}"
f"/{workflow_id}/{organization_id}/{workflow_run_id}"
)
# Build the webhook URL for status callbacks
@ -267,7 +267,7 @@ class TelnyxProvider(TelephonyProvider):
return False
async def get_webhook_response(
self, workflow_id: int, user_id: int, workflow_run_id: int
self, workflow_id: int, organization_id: int, workflow_run_id: int
) -> str:
"""Not used for Telnyx — streaming is inline with the dial request."""
return ""
@ -338,7 +338,7 @@ class TelnyxProvider(TelephonyProvider):
self,
websocket: "WebSocket",
workflow_id: int,
user_id: int,
organization_id: int,
workflow_run_id: int,
) -> None:
"""Handle Telnyx WebSocket connection for real-time audio.
@ -406,7 +406,7 @@ class TelnyxProvider(TelephonyProvider):
provider_name=self.PROVIDER_NAME,
workflow_id=workflow_id,
workflow_run_id=workflow_run_id,
user_id=user_id,
organization_id=organization_id,
call_id=call_control_id,
transport_kwargs={
"stream_id": stream_id,

8
api/services/telephony/providers/twilio/provider.py
View file

@ -166,7 +166,7 @@ class TwilioProvider(TelephonyProvider):
return validator.validate(url, params, signature)
async def get_webhook_response(
self, workflow_id: int, user_id: int, workflow_run_id: int
self, workflow_id: int, organization_id: int, workflow_run_id: int
) -> str:
"""
Generate TwiML response for starting a call session.
@ -176,7 +176,7 @@ class TwilioProvider(TelephonyProvider):
twiml_content = f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Connect>
<Stream url="{wss_backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{user_id}/{workflow_run_id}"></Stream>
<Stream url="{wss_backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{organization_id}/{workflow_run_id}"></Stream>
</Connect>
<Pause length="40"/>
</Response>"""
@ -274,7 +274,7 @@ class TwilioProvider(TelephonyProvider):
self,
websocket: "WebSocket",
workflow_id: int,
user_id: int,
organization_id: int,
workflow_run_id: int,
) -> None:
"""
@ -325,7 +325,7 @@ class TwilioProvider(TelephonyProvider):
provider_name=self.PROVIDER_NAME,
workflow_id=workflow_id,
workflow_run_id=workflow_run_id,
user_id=user_id,
organization_id=organization_id,
call_id=call_sid,
transport_kwargs={"stream_sid": stream_sid, "call_sid": call_sid},
)

3
api/services/telephony/providers/twilio/routes.py
View file

@ -47,7 +47,6 @@ async def _persist_amd_result_if_present(
@router.post("/twiml", include_in_schema=False)
async def handle_twiml_webhook(
workflow_id: int,
user_id: int,
workflow_run_id: int,
organization_id: int,
request: Request,
@ -79,7 +78,7 @@ async def handle_twiml_webhook(
)
response_content = await provider.get_webhook_response(
workflow_id, user_id, workflow_run_id
workflow_id, organization_id, workflow_run_id
)
return HTMLResponse(content=response_content, media_type="application/xml")

8
api/services/telephony/providers/vobiz/provider.py
View file

@ -255,7 +255,7 @@ class VobizProvider(TelephonyProvider):
return is_valid
async def get_webhook_response(
self, workflow_id: int, user_id: int, workflow_run_id: int
self, workflow_id: int, organization_id: int, workflow_run_id: int
) -> str:
"""
Generate Vobiz XML response for starting a call session.
@ -269,7 +269,7 @@ class VobizProvider(TelephonyProvider):
vobiz_xml = f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Stream bidirectional="true" keepCallAlive="true" contentType="audio/x-mulaw;rate=8000">{wss_backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{user_id}/{workflow_run_id}</Stream>
<Stream bidirectional="true" keepCallAlive="true" contentType="audio/x-mulaw;rate=8000">{wss_backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{organization_id}/{workflow_run_id}</Stream>
</Response>"""
return vobiz_xml
@ -350,7 +350,7 @@ class VobizProvider(TelephonyProvider):
self,
websocket: "WebSocket",
workflow_id: int,
user_id: int,
organization_id: int,
workflow_run_id: int,
) -> None:
"""
@ -394,7 +394,7 @@ class VobizProvider(TelephonyProvider):
provider_name=self.PROVIDER_NAME,
workflow_id=workflow_id,
workflow_run_id=workflow_run_id,
user_id=user_id,
organization_id=organization_id,
call_id=call_id,
transport_kwargs={"stream_id": stream_id, "call_id": call_id},
)

6
api/services/telephony/providers/vobiz/routes.py
View file

@ -54,7 +54,7 @@ async def _verify_vobiz_callback(
@router.post("/vobiz-xml", include_in_schema=False)
async def handle_vobiz_xml_webhook(
workflow_id: int, user_id: int, workflow_run_id: int, organization_id: int
workflow_id: int, workflow_run_id: int, organization_id: int
):
"""
Handle initial webhook from Vobiz when call is answered.
@ -65,7 +65,7 @@ async def handle_vobiz_xml_webhook(
set_current_run_id(workflow_run_id)
logger.info(
f"[run {workflow_run_id}] Vobiz XML webhook called - "
f"workflow_id={workflow_id}, user_id={user_id}, org_id={organization_id}"
f"workflow_id={workflow_id}, org_id={organization_id}"
)
workflow_run = await db_client.get_workflow_run_by_id(workflow_run_id)
@ -74,7 +74,7 @@ async def handle_vobiz_xml_webhook(
logger.debug(f"[run {workflow_run_id}] Using provider: {provider.PROVIDER_NAME}")
response_content = await provider.get_webhook_response(
workflow_id, user_id, workflow_run_id
workflow_id, organization_id, workflow_run_id
)
logger.debug(

17
api/services/telephony/providers/vonage/provider.py
View file

@ -207,7 +207,7 @@ class VonageProvider(TelephonyProvider):
return False
async def get_webhook_response(
self, workflow_id: int, user_id: int, workflow_run_id: int
self, workflow_id: int, organization_id: int, workflow_run_id: int
) -> str:
"""
Generate NCCO response for starting a call session.
@ -222,7 +222,7 @@ class VonageProvider(TelephonyProvider):
"endpoint": [
{
"type": "websocket",
"uri": f"{wss_backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{user_id}/{workflow_run_id}",
"uri": f"{wss_backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{organization_id}/{workflow_run_id}",
"content-type": "audio/l16;rate=16000", # 16kHz Linear PCM
"headers": {},
}
@ -323,7 +323,7 @@ class VonageProvider(TelephonyProvider):
self,
websocket: "WebSocket",
workflow_id: int,
user_id: int,
organization_id: int,
workflow_run_id: int,
) -> None:
"""
@ -338,14 +338,17 @@ class VonageProvider(TelephonyProvider):
try:
# Get workflow run to extract call UUID
workflow_run = await db_client.get_workflow_run(workflow_run_id)
workflow_run = await db_client.get_workflow_run(
workflow_run_id, organization_id=organization_id
)
if not workflow_run:
logger.error(f"Workflow run {workflow_run_id} not found")
await websocket.close(code=4404, reason="Workflow run not found")
return
# Get workflow for organization info
workflow = await db_client.get_workflow(workflow_id, user_id)
workflow = await db_client.get_workflow(
workflow_id, organization_id=organization_id
)
if not workflow:
logger.error(f"Workflow {workflow_id} not found")
await websocket.close(code=4404, reason="Workflow not found")
@ -392,7 +395,7 @@ class VonageProvider(TelephonyProvider):
provider_name=self.PROVIDER_NAME,
workflow_id=workflow_id,
workflow_run_id=workflow_run_id,
user_id=user_id,
organization_id=organization_id,
call_id=call_uuid,
transport_kwargs={"call_uuid": call_uuid},
)

10
api/services/telephony/providers/vonage/routes.py
View file

@ -5,7 +5,6 @@ provider registry — see ProviderSpec.router.
"""
import json
from typing import Optional
from fastapi import APIRouter, HTTPException, Request
from loguru import logger
@ -20,9 +19,8 @@ router = APIRouter()
@router.get("/ncco", include_in_schema=False)
async def handle_ncco_webhook(
workflow_id: int,
user_id: int,
workflow_run_id: int,
organization_id: Optional[int] = None,
organization_id: int,
):
"""Handle NCCO (Nexmo Call Control Objects) webhook for Vonage.
@ -30,12 +28,10 @@ async def handle_ncco_webhook(
"""
workflow_run = await db_client.get_workflow_run_by_id(workflow_run_id)
provider = await get_telephony_provider_for_run(
workflow_run, organization_id or user_id
)
provider = await get_telephony_provider_for_run(workflow_run, organization_id)
response_content = await provider.get_webhook_response(
workflow_id, user_id, workflow_run_id
workflow_id, organization_id, workflow_run_id
)
return json.loads(response_content)

2
api/tests/telephony/cloudonix/test_routes.py
View file

@ -94,7 +94,6 @@ async def test_agent_stream_reads_call_metadata_from_start_message():
websocket,
organization_id=44,
workflow_id=101,
user_id=202,
workflow_run_id=303,
params={},
)
@ -267,7 +266,6 @@ async def test_agent_stream_handshake_timeout_closes_socket(monkeypatch):
websocket,
organization_id=1,
workflow_id=2,
user_id=3,
workflow_run_id=4,
params={},
),

4
api/tests/telephony/plivo/test_routes.py
View file

@ -89,7 +89,6 @@ async def test_plivo_xml_route_accepts_valid_signature_with_extra_query_param():
provider = _provider()
query = {
"workflow_id": 7,
"user_id": 8,
"workflow_run_id": 123,
"campaign_id": 42,
"organization_id": 11,
@ -138,14 +137,13 @@ async def test_plivo_xml_route_accepts_valid_signature_with_extra_query_param():
response = await handle_plivo_xml_webhook(
workflow_id=7,
user_id=8,
workflow_run_id=123,
organization_id=11,
request=request,
)
assert response.body == b"<Response/>"
get_webhook_response.assert_awaited_once_with(7, 8, 123)
get_webhook_response.assert_awaited_once_with(7, 11, 123)
db_client.update_workflow_run.assert_awaited_once()

6
api/tests/telephony/twilio/test_routes.py
View file

@ -109,7 +109,6 @@ async def test_twiml_route_accepts_valid_signature_with_extra_query_param():
provider = _provider()
query = {
"workflow_id": 7,
"user_id": 8,
"workflow_run_id": 123,
"campaign_id": 42,
"organization_id": 11,
@ -149,14 +148,13 @@ async def test_twiml_route_accepts_valid_signature_with_extra_query_param():
response = await handle_twiml_webhook(
workflow_id=7,
user_id=8,
workflow_run_id=123,
organization_id=11,
request=request,
)
assert response.body == b"<Response/>"
get_webhook_response.assert_awaited_once_with(7, 8, 123)
get_webhook_response.assert_awaited_once_with(7, 11, 123)
@pytest.mark.asyncio
@ -166,7 +164,6 @@ async def test_twiml_route_rejects_missing_signature():
path="/api/v1/telephony/twiml",
query={
"workflow_id": 7,
"user_id": 8,
"workflow_run_id": 123,
"organization_id": 11,
},
@ -188,7 +185,6 @@ async def test_twiml_route_rejects_missing_signature():
with pytest.raises(HTTPException) as exc_info:
await handle_twiml_webhook(
workflow_id=7,
user_id=8,
workflow_run_id=123,
organization_id=11,
request=request,

2
api/tests/test_active_calls.py
View file

@ -183,7 +183,7 @@ async def test_telephony_entrypoint_counts_call_during_setup(monkeypatch):
provider_name="twilio",
workflow_id=1,
workflow_run_id=44,
user_id=7,
organization_id=7,
call_id="call-1",
transport_kwargs={},
)

3
api/tests/test_public_agent_routes.py
View file

@ -126,7 +126,8 @@ def test_trigger_route_executes_as_workflow_owner():
initiate_kwargs = provider.initiate_call.await_args.kwargs
assert initiate_kwargs["workflow_id"] == workflow.id
assert initiate_kwargs["user_id"] == workflow.user_id
# The media websocket URL is keyed on the org, not the workflow owner.
assert initiate_kwargs["organization_id"] == workflow.organization_id
def test_workflow_uuid_route_uses_scoped_lookup_and_shared_execution():

14
api/tests/test_telephony_routes.py
View file

@ -123,8 +123,12 @@ def test_initiate_call_executes_as_workflow_owner_for_shared_org_workflow():
initiate_kwargs = provider.initiate_call.await_args.kwargs
assert initiate_kwargs["workflow_id"] == workflow.id
assert initiate_kwargs["user_id"] == workflow.user_id
assert "user_id=99" in initiate_kwargs["webhook_url"]
# The media websocket URL is keyed on the org, not the workflow owner.
assert initiate_kwargs["organization_id"] == workflow.organization_id
webhook_url = initiate_kwargs["webhook_url"]
assert f"organization_id={workflow.organization_id}" in webhook_url
# The answer URL carries no workflow owner: nothing downstream scopes on it.
assert "user_id=" not in webhook_url
mock_db.get_user_configurations.assert_not_called()
@ -380,11 +384,13 @@ async def test_smallwebrtc_run_reaching_telephony_websocket_closes_without_runni
):
mock_concurrency.unregister_active_call = AsyncMock()
mock_db.get_workflow_run = AsyncMock(return_value=workflow_run)
mock_db.get_workflow_by_id = AsyncMock(return_value=workflow)
mock_db.get_workflow = AsyncMock(return_value=workflow)
mock_db.update_workflow_run = AsyncMock()
await _handle_telephony_websocket(websocket, 33, 99, 501)
await _handle_telephony_websocket(websocket, 33, 11, 501)
mock_db.get_workflow_run.assert_awaited_once_with(501, organization_id=11)
mock_db.get_workflow.assert_awaited_once_with(33, organization_id=11)
websocket.close.assert_awaited_once_with(
code=4400,
reason=(

2
docs/integrations/telephony/asterisk-ari.mdx
View file

@ -107,7 +107,7 @@ The section name (e.g., `dograh`) is the **WebSocket Client Name** you'll enter
</Note>
<Note>
Configure the `uri` as a base URL only, without a query string. During each call, Dograh asks Asterisk to create an `externalMedia` channel and Asterisk appends `workflow_id`, `user_id`, and `workflow_run_id` through the `v()` transport data for that call. Opening `/api/v1/telephony/ws/ari` directly in a browser or with `wscat` can return HTTP 403 because those routing parameters are missing; that is expected and does not indicate a `websocket_client.conf` misconfiguration.
Configure the `uri` as a base URL only, without a query string. During each call, Dograh asks Asterisk to create an `externalMedia` channel and Asterisk appends `workflow_id`, `organization_id`, and `workflow_run_id` through the `v()` transport data for that call. Opening `/api/v1/telephony/ws/ari` directly in a browser or with `wscat` can return HTTP 403 because those routing parameters are missing; that is expected and does not indicate a `websocket_client.conf` misconfiguration.
</Note>
<Note>

14
docs/integrations/telephony/custom.mdx
View file

@ -71,11 +71,11 @@ class YourProvider(TelephonyProvider):
# ---------- webhooks ----------
async def verify_webhook_signature(self, url, params, signature) -> bool: ...
async def get_webhook_response(self, workflow_id, user_id, workflow_run_id) -> str: ...
async def get_webhook_response(self, workflow_id, organization_id, workflow_run_id) -> str: ...
def parse_status_callback(self, data: dict) -> dict: ...
# ---------- websocket ----------
async def handle_websocket(self, websocket, workflow_id, user_id, workflow_run_id): ...
async def handle_websocket(self, websocket, workflow_id, organization_id, workflow_run_id): ...
# ---------- inbound ----------
@classmethod
@ -112,6 +112,14 @@ class YourProvider(TelephonyProvider):
See `api/services/telephony/base.py` for the full docstrings on each method.
<Warning>
`get_webhook_response` and `handle_websocket` receive an `organization_id`, not a
user id. It is the tenant that every workflow and workflow-run lookup must be
scoped by — pass it straight through to `run_pipeline_telephony`. The workflow
owner is deliberately not handed to providers; read it off the workflow row if you
need it for attribution.
</Warning>
## Implementation Guide
### 1. Configuration schemas
@ -385,7 +393,7 @@ For end-to-end testing, save your provider through the telephony-configurations
1. **Trust the registry** — never import another provider's class directly; resolve through the factory helpers (`get_default_telephony_provider`, `get_telephony_provider_by_id`, etc.).
2. **Sensitive fields** — mark every credential field `sensitive=True` in `ProviderUIMetadata`. The save endpoint masks these on read and preserves the original when the client re-submits a masked value.
3. **Inbound signature verification** — always validate inbound webhook signatures in `verify_inbound_signature`. Returning `True` when no signature header is present is acceptable; return `False` when a signature *is* present but invalid.
3. **Inbound signature verification** — always validate inbound webhook signatures in `verify_inbound_signature`, and fail closed. If your provider signs every webhook (most do), a *missing* signature header means the request did not come from the provider: return `False`, exactly as `providers/twilio/` and `providers/plivo/` do. Never `return True` as a placeholder — these handlers are reachable by anyone who can guess a `workflow_run_id`.
4. **Transports load credentials lazily** — call `load_credentials_for_transport` with the `telephony_configuration_id` from the workflow run. Don't read the org's default config from `transport.py`.
5. **Logging** — use `loguru.logger`.

71
docs/integrations/telephony/webhooks.mdx
View file

@ -5,17 +5,28 @@ description: "How Dograh AI handles telephony webhooks and audio streaming"
## Overview
Dograh AI uses webhooks to communicate with telephony providers for call events and audio streaming. Webhooks are automatically configured when you initiate calls.
Dograh AI uses webhooks to communicate with telephony providers for call events and audio streaming. For outbound calls, Dograh builds these URLs and hands them to the provider when it dials — you never configure them by hand. For inbound calls, you point your provider at a single dispatcher URL; see [Inbound Calls](/integrations/telephony/inbound).
Every webhook path lives under `/api/v1/telephony`.
## Webhook Types
### 1. Initial Call Webhook
### 1. Answer Webhook
When a call is initiated, the telephony provider requests instructions.
When an outbound call connects, the provider requests instructions. The path is provider-specific, and Dograh appends the routing parameters as a query string:
**Endpoint**: `/api/v1/telephony/webhook/{workflow_id}/{user_id}/{workflow_run_id}`
```
?workflow_id={workflow_id}&workflow_run_id={workflow_run_id}&organization_id={organization_id}
```
**Purpose**: Returns provider-specific instructions
| Provider | Method | Path |
| --- | --- | --- |
| Twilio, Cloudonix | `POST` | `/twiml` |
| Plivo | `POST` | `/plivo-xml` |
| Vobiz | `POST` | `/vobiz-xml` |
| Vonage | `GET` | `/ncco` |
Telnyx and Asterisk ARI have no answer webhook. Telnyx is call-control style — Dograh POSTs the stream and event URLs to Telnyx's API instead of returning markup. ARI streams over a WebSocket only.
<Tabs>
<Tab title="Twilio (TwiML)">
@ -23,7 +34,7 @@ When a call is initiated, the telephony provider requests instructions.
<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Connect>
<Stream url="wss://your-domain/api/v1/telephony/ws/123/456/789" />
<Stream url="wss://your-domain/api/v1/telephony/ws/123/11/789" />
</Connect>
</Response>
```
@ -35,7 +46,7 @@ When a call is initiated, the telephony provider requests instructions.
"action": "connect",
"endpoint": [{
"type": "websocket",
"uri": "wss://your-domain/api/v1/telephony/ws/123/456/789",
"uri": "wss://your-domain/api/v1/telephony/ws/123/11/789",
"content-type": "audio/l16;rate=16000"
}]
}
@ -44,30 +55,50 @@ When a call is initiated, the telephony provider requests instructions.
</Tab>
</Tabs>
### 2. Status Callback
Here `123` is the workflow, `11` the organization, and `789` the workflow run.
Receives call lifecycle events.
### 2. Status Callbacks
**Endpoint**: `/api/v1/telephony/status-callback/{workflow_run_id}`
Receive call lifecycle events. Each is keyed on the workflow run:
| Provider | Path |
| --- | --- |
| Twilio | `/twilio/status-callback/{workflow_run_id}` |
| Plivo | `/plivo/hangup-callback/{workflow_run_id}`, `/plivo/ring-callback/{workflow_run_id}` |
| Vobiz | `/vobiz/hangup-callback/{workflow_run_id}`, `/vobiz/ring-callback/{workflow_run_id}` |
| Vonage | `/vonage/events/{workflow_run_id}` |
| Telnyx | `/telnyx/events/{workflow_run_id}` |
| Cloudonix | `/cloudonix/status-callback/{workflow_run_id}`, `/cloudonix/cdr` |
Providers report their own vocabulary; Dograh normalizes it into a common set of states:
**Events Tracked**:
- `initiated` - Call request received
- `ringing` - Call is ringing
- `in-progress` - Call is connected and streaming
- `answered` - Call was answered
- `completed` - Call ended normally
- `busy` - Line was busy
- `no-answer` - Call not answered
- `canceled` - Call was canceled before connecting
- `failed` - Call failed
- `error` - Provider reported an error
A status Dograh does not recognize is passed through unchanged rather than dropped.
### 3. WebSocket Audio Stream
Real-time audio streaming for voice interaction.
**Endpoint**: `/api/v1/telephony/ws/{workflow_id}/{user_id}/{workflow_run_id}`
**Endpoint**: `/api/v1/telephony/ws/{workflow_id}/{organization_id}/{workflow_run_id}`
Asterisk ARI instead connects to `/api/v1/telephony/ws/ari` and passes the same three values as query parameters — see [Asterisk ARI](/integrations/telephony/asterisk-ari).
The `organization_id` segment is the tenant that owns the workflow. Dograh scopes every workflow and workflow-run lookup by it, so a run belonging to one organization can never be served under another's id.
**Audio Formats**:
- **Twilio**: 8kHz μ-law (MULAW), Base64-encoded in JSON messages
- **Twilio / Plivo / Vobiz**: 8kHz μ-law (MULAW), Base64-encoded in JSON messages
- **Vonage**: 16kHz Linear PCM, Binary frames
- **Asterisk ARI**: 8kHz Linear PCM via externalMedia
## How It Works
@ -76,11 +107,14 @@ Dograh AI automatically:
2. Passes them to the telephony provider when initiating calls
3. Verifies webhook signatures for security:
- **Twilio**: HMAC-SHA1 signature validation
- **Plivo / Vobiz**: HMAC-SHA256 signature validation
- **Vonage**: JWT token verification
4. Processes status updates to track call lifecycle
5. Manages WebSocket connections for audio streaming
6. Handles provider-specific audio formats and protocols
Signature verification is implemented per provider in `verify_inbound_signature`. If you are adding a provider, see [Custom Telephony Provider](/integrations/telephony/custom).
## Local Development
For local development, use the built-in Cloudflare tunnel:
@ -102,16 +136,21 @@ The tunnel URL is automatically detected and used for webhooks.
- Check firewall rules allow incoming HTTPS traffic
- Test with `curl` from external network
</Accordion>
<Accordion title="Signature verification failures">
- Providers sign the full URL, query string included — a proxy that rewrites or reorders query parameters will invalidate the signature
- Confirm the backend's public URL matches what the provider was given, including scheme and port
</Accordion>
<Accordion title="WebSocket connection dropping">
- Check WebSocket upgrade headers are preserved
- Verify no timeout on load balancer/proxy
- Monitor for memory/CPU constraints
</Accordion>
<Accordion title="Status callbacks not received">
- Verify workflow_run_id is included in URL
- Check provider console for webhook errors
- Review webhook retry logs
</Accordion>
</AccordionGroup>
</AccordionGroup>

7
ui/src/client/types.gen.ts
View file

@ -6121,12 +6121,7 @@ export type UpdateWorkflowRequest = {
template_context_variables?: {
[key: string]: unknown;
} | null;
/**
* Workflow Configurations
*/
workflow_configurations?: {
[key: string]: unknown;
} | null;
workflow_configurations?: WorkflowConfigurationDefaults | null;
};
/**