AgentsFunctionResult

tap

View as MarkdownOpen in Claude

Start a media tap that streams call audio to an external endpoint. Supports WebSocket (wss://, ws://) and RTP (rtp://) destinations.

Raises ValueError if direction is not "speak", "listen", or "both", if codec is not "PCMU" or "PCMA", or if rtp_ptime is not a positive integer.

Parameters

uri
strRequired

Destination URI for the media stream. Supported formats:

  • "rtp://IP:port" — RTP stream
  • "ws://example.com" — WebSocket stream
  • "wss://example.com" — Secure WebSocket stream
control_id
str | NoneDefaults to None

Identifier for this tap. Pass the same ID to stop_tap() to end this specific tap. If omitted, a default ID is generated.

direction
Literal['speak', 'listen', 'both']Defaults to both

Audio direction to tap.

  • "speak" — what the party says
  • "listen" — what the party hears
  • "both" — both directions
codec
Literal['PCMU', 'PCMA']Defaults to PCMU

Audio codec for the stream.

  • "PCMU" — G.711 mu-law
  • "PCMA" — G.711 A-law
rtp_ptime
intDefaults to 20

Packetization time in milliseconds for RTP streams. Must be a positive integer.

status_url
str | NoneDefaults to None

URL to receive tap status change webhooks.

Returns

FunctionResult — self, for chaining.

Example

from signalwire import AgentBase
from signalwire import FunctionResult
agent = AgentBase(name="my-agent", route="/agent")
agent.set_prompt_text("You are a helpful assistant.")
@agent.tool(name="start_monitoring", description="Start monitoring the call audio")
def start_monitoring(args, raw_data):
    return (
        FunctionResult("Call monitoring started.")
        .tap(
            uri="wss://monitor.example.com/audio",
            control_id="supervisor_tap",
            direction="both"
        )
    )
agent.serve()