Server API Reference

Reference for the primary Python APIs used to run Piopiy voice agents.

Agent

Agent connects to signaling and creates one task per active room/session.

Constructor

Agent(
    agent_id: str,
    agent_token: str,
    create_session: Callable[..., Awaitable[None]],
    signaling_url: Optional[str] = None,
    debug: bool = False,
)

Parameter	Type	Required	Description
agent_id	str	Yes	Piopiy agent ID from dashboard.
agent_token	str	Yes	Piopiy agent token from dashboard.
create_session	Callable[..., Awaitable[None]]	Yes	Async callback invoked per session.
signaling_url	str | None	No	Custom signaling endpoint override.
debug	bool	No	INFO logs when True, quieter logs when False.

Methods

await agent.connect()
await agent.shutdown()

• connect() opens signaling connection and waits for events.
• shutdown() disconnects and cancels active session tasks.

Session Callback Contract

Agent inspects your callback signature and only passes supported fields when present:

• agent_id
• call_id
• from_number
• to_number
• metadata

metadata is parsed to JSON object when possible.

Context Variables

From piopiy.agent:

• URL_CTX
• TOKEN_CTX
• ROOM_CTX

These contextvars are set per running session.

VoiceAgent

VoiceAgent builds and runs the STT -> LLM -> TTS pipeline.

Constructor

VoiceAgent(
    *,
    instructions: str,
    tools: Optional[List[FunctionSchema]] = None,
    greeting: Optional[str] = None,
    idle_timeout_secs: int = 60,
)

Parameter	Type	Required	Description
instructions	str	Yes	System-level behavior prompt.
tools	List[FunctionSchema] | None	No	Optional initial tool schemas.
greeting	str | None	No	Spoken greeting after participant joins.
idle_timeout_secs	int	No	Idle timeout for cancellation behavior.

Action(...)

await voice_agent.Action(
    *,
    stt: Optional[FrameProcessor] = None,
    llm: FrameProcessor,
    tts: Optional[FrameProcessor] = None,
    tts_switcher: Optional[ServiceSwitcher] = None,
    stt_switcher: Optional[ServiceSwitcher] = None,
    mcp_tools: Optional[Any] = None,
    vad: Optional[Any] = None,
    enable_metrics: bool = True,
    enable_usage_metrics: bool = True,
    allow_interruptions: bool = True,
    interruption_strategy: Optional[BaseInterruptionStrategy] = None,
    telecmi_params: Optional[TelecmiParams] = None,
)

Runtime requirements:

• llm is required.
• Provide stt or stt_switcher.
• Provide tts or tts_switcher.

start()

await voice_agent.start()

Builds task/pipeline (if needed) and runs until call end/cancel.

Tool APIs

voice_agent.add_tool(schema, handler)
voice_agent.register_tool(name, handler)

Service Switch API

await voice_agent.switch_service(service)

Queues a manual switch frame for switcher-managed services.

VAD Input Modes

vad accepts multiple modes:

• True: enable default Silero analyzer.
• dict: mapped keys confidence, start_secs, stop_secs, min_volume.
• SileroVADAnalyzer or compatible processor instance.
• False or None: disabled.

ServiceSwitcher

Import and initialize manual strategy:

from piopiy.pipeline.service_switcher import ServiceSwitcher, ServiceSwitcherStrategyManual

switcher = ServiceSwitcher(
    services=[service_a, service_b],
    strategy_type=ServiceSwitcherStrategyManual,
)

Related Docs

• Telephony
• Ecosystem
• Quickstart