tap

Intercept call media (audio) and stream it to an external destination such as a WebSocket or RTP endpoint. Returns a TapAction that you can use to stop the tap or wait for it to finish.

This method emits calling.call.tap events. See Call Events for payload details.

Parameters

tap

dictRequired

Tap configuration specifying which audio to intercept.

tap.type

strRequired

Tap type. Valid values: "audio".

tap.params

dict

Tap parameters. Supports a direction key with the following values:

"listen" — capture audio heard by the caller
"speak" — capture audio spoken by the caller
"both" — capture audio in both directions

device

dictRequired

Destination device for the tapped media.

device.type

strRequired

Device type.

"ws" — WebSocket endpoint
"rtp" — RTP endpoint

device.params

dictRequired

Device-specific parameters (e.g., uri for WebSocket, addr/port for RTP).

control_id

Optional[str]

Custom control ID. Auto-generated if not provided.

on_completed

Optional[Callable[[RelayEvent], Any]]

Callback invoked when the tap operation ends.

Returns

TapAction — An action handle with stop() and wait() methods.

Example

1 import asyncio
2 from signalwire.relay import RelayClient
3 
4 client = RelayClient(
5     project="your-project-id",
6     token="your-api-token",
7     host="your-space.signalwire.com",
8     contexts=["default"],
9 )
10 
11 @client.on_call
12 async def handle_call(call):
13     await call.answer()
14 
15     # Tap audio and stream to a WebSocket endpoint
16     action = await call.tap(
17         tap={"type": "audio", "params": {"direction": "both"}},
18         device={"type": "ws", "params": {"uri": "wss://example.com/tap"}},
19     )
20 
21     # Tap runs in background; stop it later
22     await asyncio.sleep(30)
23     await action.stop()
24 
25 client.run()

1	import asyncio
2	from signalwire.relay import RelayClient
3
4	client = RelayClient(
5	project="your-project-id",
6	token="your-api-token",
7	host="your-space.signalwire.com",
8	contexts=["default"],
9	)
10
11	@client.on_call
12	async def handle_call(call):
13	await call.answer()
14
15	# Tap audio and stream to a WebSocket endpoint
16	action = await call.tap(
17	tap={"type": "audio", "params": {"direction": "both"}},
18	device={"type": "ws", "params": {"uri": "wss://example.com/tap"}},
19	)
20
21	# Tap runs in background; stop it later
22	await asyncio.sleep(30)
23	await action.stop()
24
25	client.run()