The claim The bundled schema describes tap’s uri as the “destination of the tap media stream: rtp://IP:port, ws://example.com, or wss://example.com”. direction is speak “for what party says”, listen “for what party hears”, or both. codec is PCMU or PCMA. control_id is the “identifier for this tap to use with stop_tap”. The bundled schema requires only uri. The vendored REST spec has the same operation as the call command calling.tap. It takes a tap configuration of type audio and a device of type ws or rtp, and calling.tap.stop ends it. The SDK wraps those as client.calling.tap and client.calling.tap_stop.
How it works - answer: {}
- tap:
uri: "wss://media.example.com/tap"
control_id: "workshop-tap"
direction: "both"
codec: "PCMU"
- connect:
to: "+15550100001"
timeout: 20
- stop_tap:
control_id: "workshop-tap"
- hangup: {} The document carries tap for both directions, then connect, then stop_tap with the same control id. The Python surface builds the same document with SWMLService and adds the REST pair:
def start_tap(call_id):
return client.calling.tap(call_id, control_id=CONTROL_ID,
tap={"type": "audio", "params": {"direction": "both"}},
device={"type": "ws", "params": {"uri": TAP_URI}})
def stop_tap(call_id):
return client.calling.tap_stop(call_id, control_id=CONTROL_ID) What the platform receives from start_tap:
{"command": "calling.tap", "id": "6d3f4a0e-2b1c-4e7a-9f0d-1c2b3a4d5e6f",
"params": {"control_id": "workshop-tap",
"tap": {"type": "audio", "params": {"direction": "both"}},
"device": {"type": "ws", "params": {"uri": "wss://media.example.com/tap"}}}} The spec describes the RTP device’s addr as a “public IPv4 address” and says “private/reserved ranges are rejected”. It says the WebSocket uri “must start with ws:// or wss://”.
Limitations You prove the documents and the requests. What arrives on your socket, and in what framing, is the platform’s side of a live call and belongs to the tap documentation.
What to change first Change direction to speak in both surfaces and run the verifier. The tap equality fails, which is the point: speak, listen and both are the three choices, and the document names one.
Markup Python
swml/agent.yaml python/app.py Copy
# Stream call audio to your own server. `tap` forks the call's audio to a
# WebSocket or RTP destination while the call carries on; `stop_tap` ends the
# fork by its control_id. Here both sides of the conversation stream to a
# wss:// server while the caller talks to the workshop, then the tap stops.
version : "1.0.0"
sections :
main :
- answer : {}
- tap :
uri : "wss://media.example.com/tap"
control_id : "workshop-tap"
direction : "both"
codec : "PCMU"
- connect :
to : "+15550100001"
timeout : 20
- stop_tap :
control_id : "workshop-tap"
- hangup : {} """Stream call audio to your own server.
`tap` sends a copy of the call's audio to a destination of yours. The bundled
schema gives it a `uri` of `rtp://IP:port`, `ws://` or `wss://` and a
`direction` of `speak`, `listen` or `both`. `codec` is PCMU or PCMA, and
`stop_tap` ends the tap by its `control_id`. Over REST the same operation is
the call command `calling.tap`, with a `device` of type `ws` or `rtp`.
Written against signalwire-sdk 3.0.1 (SWMLService, RestClient.calling).
"""
import os
from dotenv import load_dotenv
from signalwire import SWMLService
from signalwire.rest import RestClient
# the SDK does not read .env for you
load_dotenv ()
# RestClient() reads SIGNALWIRE_PROJECT_ID / SIGNALWIRE_API_TOKEN /
# SIGNALWIRE_SPACE from the environment (signalwire/rest/client.py).
client = RestClient ()
TAP_URI = os . getenv ( "TAP_URI" , "wss://media.example.com/tap" )
OWNER = os . getenv ( "OWNER_NUMBER" , "+15550100001" )
CONTROL_ID = "workshop-tap"
def build ( service = None ):
"""Fork both directions to your server for the length of the bridge."""
service = service or SWMLService ( name = "tap" , route = "/tap" )
service . reset_document ()
service . add_verb ( "answer" , {})
service . add_verb ( "tap" , { "uri" : TAP_URI , "control_id" : CONTROL_ID ,
"direction" : "both" , "codec" : "PCMU" })
service . add_verb ( "connect" , { "to" : OWNER , "timeout" : 20 })
service . add_verb ( "stop_tap" , { "control_id" : CONTROL_ID })
service . add_verb ( "hangup" , {})
return service
def start_tap ( call_id ):
"""The same fork, started mid-call from your backend."""
return client . calling . tap ( call_id , control_id = CONTROL_ID ,
tap = { "type" : "audio" , "params" : { "direction" : "both" }},
device = { "type" : "ws" , "params" : { "uri" : TAP_URI }})
def stop_tap ( call_id ):
return client . calling . tap_stop ( call_id , control_id = CONTROL_ID )
if __name__ == "__main__" :
build () . serve ( port = int ( os . getenv ( "PORT" , "8080" ))) Run it cd python
pip install -r requirements.txt
cp ../.env.example .env # then edit .env: TAP_URI, OWNER_NUMBER, credentials, basic-auth pair
python app.py TAP_URI is a WebSocket server you run; the schema calls it the destination of the tap media stream. The webhook needs a public HTTPS URL. For a local run, expose port 8080 with a tunnel such as ngrok and use that hostname. Point a number’s SWML webhook at https://<user>:<password>@<your-host>/tap/ and call. To fork a call that is already up, run python -c "from app import start_tap; print(start_tap('<call_id>'))" from the python/ folder.
Verify it No network, no account.
cd .. # back to the recipe folder
python verify.py When you run it, you validate both surfaces, swap the SDK’s HTTP layer for a recorder, and assert the following.
both surfaces contain answer, tap, connect, stop_tap, hangup in order and render the same document the tap object is exactly the wss URI, both, PCMU and the control id, and stop_tap repeats that id the bundled schema requires uri on tap and nothing else start_tap and stop_tap each make one POST to the documented calling path, and each body equals one expected objectthe vendored REST spec documents every calling.tap param and its nested tap and ws device, including their required fields and the type and direction enums the calling.tap.stop params are exactly the control id, and the vendored REST spec documents them and requires nothing more