RELAYCall

tap

View as MarkdownOpen in Claude

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

1import asyncio
2from signalwire.relay import RelayClient
3
4client = 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
12async 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
25client.run()