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

fix: fix agent stream contract with cloudonix (#504)

Browse source
This commit is contained in:
Abhishek 2026-07-06 23:58:04 +05:30 • committed by GitHub
parent 7e2750f83f
commit 5a822232da
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 319 additions and 114 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -776,7 +776,9 @@ class ARIConnection:
bridge_id = ctx.get("bridge_id")
transfer_state = ctx.get("transfer_state")
transfer_bridge_id = ctx.get("transfer_bridge_id") or bridge_id
transfer_caller_channel_id = ctx.get("transfer_caller_channel_id") or call_id
transfer_caller_channel_id = (
ctx.get("transfer_caller_channel_id") or call_id
)
transfer_destination_channel_id = ctx.get("transfer_destination_channel_id")
# Check if this is a call transfer scenario external channel. Skip full teardown if
@ -813,7 +815,8 @@ class ARIConnection:
and transfer_bridge_id
and transfer_caller_channel_id
and transfer_destination_channel_id
and channel_id in (
and channel_id
in (
transfer_caller_channel_id,
transfer_destination_channel_id,
)
@ -884,9 +887,7 @@ class ARIConnection:
# Clean up the Redis marker for external channel
await self._delete_ext_channel(ext_channel_id)
await self._delete_transfer_channel_mapping(
transfer_destination_channel_id
)
await self._delete_transfer_channel_mapping(transfer_destination_channel_id)
logger.info(
f"[ARI org={self.organization_id}] StasisEnd full teardown for "

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

@ -359,13 +359,12 @@ class TelephonyProvider(ABC):
workflow_run_id: int,
params: Dict[str, str],
) -> None:
"""Handle the agent-stream WebSocket where credentials are passed inline.
"""Handle the provider-specific agent-stream WebSocket.
Used by ``/api/v1/agent-stream/{workflow_uuid}`` when the caller carries
provider credentials in the query string (no stored
``TelephonyConfigurationModel`` row required). ``organization_id`` is
passed so providers can scope any config lookups to the workflow's
org. Default raises so providers that haven't opted in fail loudly.
Used by ``/api/v1/agent-stream/{provider_name}/{workflow_uuid}`` when
the caller carries a provider stream protocol. ``organization_id`` is
passed so providers can scope any config lookups to the workflow's org.
Default raises so providers that haven't opted in fail loudly.
"""
raise NotImplementedError(
f"Agent-stream not supported for provider {self.PROVIDER_NAME}"

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

@ -459,10 +459,15 @@ class CloudonixProvider(TelephonyProvider):
await websocket.close(code=4400, reason="Expected start event")
return
# Extract Twilio-compatible identifiers
start = start_msg.get("start")
if not isinstance(start, dict):
logger.error("Cloudonix start message missing start object")
await websocket.close(code=4400, reason="Missing start metadata")
return
try:
stream_sid = start_msg["start"]["streamSid"]
call_sid = start_msg["start"]["callSid"]
stream_sid = start["streamSid"]
call_sid = start["callSid"]
except KeyError:
logger.error("Missing streamSid or callSid in start message")
await websocket.close(code=4400, reason="Missing stream identifiers")
@ -512,48 +517,21 @@ class CloudonixProvider(TelephonyProvider):
) -> None:
"""Agent-stream entry point.
``Domain`` (domain id) is read from the query string. The bearer
token comes from the stored Cloudonix telephony configuration
matched by ``domain_id`` within the workflow's organization — never
from the URL or stream payload. The websocket handshake (connected
/ start) is identical to the standard inbound flow.
The Cloudonix domain is read from the ``start.accountSid`` field
in the start message. The bearer token comes from the stored
Cloudonix telephony configuration matched by ``domain_id`` within
the workflow's organization — never from the URL or stream payload.
The websocket handshake (connected / start) is identical to the
standard inbound flow.
Before starting the pipeline we (a) require an existing Cloudonix
telephony configuration for the supplied ``domain_id`` and (b)
telephony configuration for the stream's ``domain_id`` and (b)
validate the call session with Cloudonix using the bearer token
from that configuration. Either failure closes the socket with
4400.
"""
from api.services.pipecat.run_pipeline import run_pipeline_telephony
domain_id = self._normalize_domain(params.get("Domain"))
if not domain_id:
logger.error("Cloudonix agent-stream missing required param: Domain")
await websocket.close(code=4400, reason="Missing Domain query param")
return
config = await self._find_config_by_domain(organization_id, domain_id)
if not config:
logger.error(
f"Cloudonix agent-stream: no telephony configuration found "
f"for domain_id={domain_id}"
)
await websocket.close(
code=4400, reason=f"Unknown Cloudonix domain: {domain_id}"
)
return
bearer_token = (config.credentials or {}).get("bearer_token")
if not bearer_token:
logger.error(
f"Cloudonix agent-stream: telephony configuration {config.id} "
f"is missing bearer_token in credentials"
)
await websocket.close(
code=4400, reason="Cloudonix configuration missing bearer_token"
)
return
try:
first_msg = await websocket.receive_text()
msg = json.loads(first_msg)
@ -568,24 +546,100 @@ class CloudonixProvider(TelephonyProvider):
await websocket.close(code=4400, reason="Expected start event")
return
start = start_msg.get("start")
if not isinstance(start, dict):
logger.error(
"Cloudonix agent-stream start message missing start object"
)
await websocket.close(code=4400, reason="Missing start metadata")
return
try:
stream_sid = start_msg["start"]["streamSid"]
call_sid = start_msg["start"]["callSid"]
call_session = start_msg["start"]["session"]
stream_sid = start["streamSid"]
call_sid = start["callSid"]
call_session = start["session"]
domain_id = self._normalize_domain(start["accountSid"])
except KeyError:
logger.error("Missing streamSid or callSid or session in start message")
logger.error(
"Missing streamSid, callSid, session, or accountSid in start message"
)
await websocket.close(code=4400, reason="Missing stream identifiers")
return
if not domain_id:
logger.error("Cloudonix agent-stream start message missing accountSid")
await websocket.close(
code=4400, reason="Missing Cloudonix domain in start message"
)
return
config = await self._find_config_by_domain(organization_id, domain_id)
if not config:
logger.error(
f"Cloudonix agent-stream: no telephony configuration found "
f"for domain_id={domain_id}"
)
await websocket.close(
code=4400, reason=f"Unknown Cloudonix domain: {domain_id}"
)
return
bearer_token = (config.credentials or {}).get("bearer_token")
if not bearer_token:
logger.error(
f"Cloudonix agent-stream: telephony configuration {config.id} "
f"is missing bearer_token in credentials"
)
await websocket.close(
code=4400, reason="Cloudonix configuration missing bearer_token"
)
return
if not await self._validate_session(domain_id, call_session, bearer_token):
await websocket.close(
code=4400, reason="Cloudonix session validation failed"
)
return
start_context = start.get("context")
await db_client.update_workflow_run(
run_id=workflow_run_id,
initial_context={
key: value
for key, value in {
"caller_number": start.get("from"),
"called_number": start.get("to"),
"direction": (
"outbound" if start_context == "outbound-api" else "inbound"
),
"cloudonix_context": start_context,
}.items()
if value is not None
},
gathered_context={
"call_id": call_session,
"cloudonix_call_sid": call_sid,
"cloudonix_stream_sid": stream_sid,
},
logs={
"inbound_webhook": {
"domain": domain_id,
"session": call_session,
"callSid": call_sid,
"streamSid": stream_sid,
"from": start.get("from"),
"to": start.get("to"),
"context": start_context,
"tracks": start.get("tracks"),
"mediaFormat": start.get("mediaFormat"),
},
},
)
logger.info(
f"Cloudonix agent-stream connected for workflow_run "
f"{workflow_run_id} stream_sid={stream_sid} call_sid={call_sid} session={call_session}"
f"{workflow_run_id} stream_sid={stream_sid} call_sid={call_sid} "
f"session={call_session} "
f"telephony_configuration_id={config.id}"
)
@ -628,7 +682,7 @@ class CloudonixProvider(TelephonyProvider):
if response.status == 200:
return True
body = await response.text()
logger.error(
logger.warning(
f"Cloudonix session validation failed: "
f"HTTP {response.status} domain_id={domain_id} "
f"call_id={call_session} body={body}"