Native support for Plivo’s WebSocket Transport with Pipecat Cloud allows you to connect your AI agents with Plivo’s voice infrastructure. This integration enables your Pipecat bots to handle real phone calls using Plivo’s WebSocket streaming.
How It Works
Pipecat Cloud implements Plivo’s bidirectional Media Streaming protocol. While audio streams flow through WebSockets, the call session is controlled by XML responses that tell Plivo how to handle each call.
When Pipecat Cloud receives an incoming WebSocket connection from Plivo, it processes the start
message to initialize a new bot instance. All WebSocket messages are forwarded to your bot, including call information such as the caller’s number and call UUID. This allows your bot to leverage Plivo’s Voice API for advanced call control - such as recording conversations, transferring to human agents, or implementing complex call flows.
Prerequisites
Before setting up this integration, ensure you have:
- A Plivo account with voice capabilities
- A Pipecat Cloud account with a Plivo WebSocket-compatible bot
- A web server to host XML responses (we’ll show you how to set this up)
A ready-to-build example of a Plivo websockets bot with complete source code is available in the pipecat-examples repo.
Plivo Setup
To connect your Pipecat Cloud bot to Plivo’s voice network:
-
Purchase a phone number from Plivo if you haven’t already. Ensure the number has voice capabilities.
-
Retrieve your Pipecat Cloud organization name using the
pipecatcloud
CLI. This information is required when creating the XML configuration.
This command will output a list of organizations associated with your account. For example:
Organization Name
──────────────────────────────────────
Default Workspace three-random-words-randomnumber (active)
- Set up an XML server to respond to Plivo webhooks. Unlike Twilio’s TwiML Bins, Plivo requires a web server to host your XML responses.
Create a file called server.py
:
from fastapi import FastAPI, Query, HTTPException
from starlette.responses import Response
import uvicorn
app = FastAPI(title="Plivo XML Server", description="Serves XML for Plivo WebSocket streaming")
@app.get("/plivo-xml")
async def plivo_xml(
agent: str = Query(..., description="Agent name"),
org: str = Query(..., description="Organization name"),
# Optional Plivo parameters that are automatically passed by Plivo
CallUUID: str = Query(None, description="Plivo call UUID"),
From: str = Query(None, description="Caller's phone number"),
To: str = Query(None, description="Called phone number"),
):
"""
Returns XML for Plivo to start WebSocket streaming to Pipecat Cloud
Required parameters:
- agent: Your deployed agent name
- org: Your Pipecat Cloud organization name
Example: /plivo-xml?agent=my-bot&org=my-org-123
"""
# Log call details (optional - useful for debugging)
if CallUUID:
print(f"Plivo call: {From} → {To}, UUID: {CallUUID}")
# Basic validation
if not agent or not org:
raise HTTPException(status_code=400, detail="Both 'agent' and 'org' parameters are required")
xml = f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Stream bidirectional="true" keepCallAlive="true" contentType="audio/x-mulaw;rate=8000">
wss://api.pipecat.daily.co/ws/plivo?serviceHost={agent}.{org}
</Stream>
</Response>"""
return Response(content=xml, media_type="application/xml")
if __name__ == "__main__":
# Run the server on port 7860
# Use with ngrok: ngrok http 7860
uvicorn.run(app, host="0.0.0.0", port=7860)
- Install dependencies and run your XML server:
pip install fastapi uvicorn
python server.py
- Make your server publicly accessible. For testing, you can use ngrok:
This will give you a public URL like https://abc123.ngrok.io
.
When testing using ngrok, use a --subdomain
arg to keep your ngrok URL
consistent.
- Create your XML application in the Plivo dashboard:
- Navigate to Voice → XML in your Plivo dashboard
- Select Add New Application
- Enter a name for your application (e.g., “Pipecat WebSocket”)
- Set the Answer URL to your ngrok URL with agent and organization parameters:
https://abc123.ngrok.io/plivo-xml?agent=AGENT_NAME&org=ORGANIZATION_NAME
- Replace
AGENT_NAME
with your deployed bot’s name (e.g., my-first-agent
)
- Replace
ORGANIZATION_NAME
with your organization name from step 2 (e.g., three-random-words-randomnumber
)
- Set HTTP Method to
GET
- Click Create Application to create your application
- Configure your Plivo phone number to use your XML application:
- Navigate to Phone Numbers and select your phone number from the list
- Select XML Application as the Application Type
- Select your XML Application from the Plivo Application dropdown
- Click Update to apply your changes
Making and Receiving Calls
Receiving Inbound Calls
To test your integration, simply dial your Plivo phone number from any phone. The call will connect to your Pipecat Cloud bot, which will respond according to your bot’s configuration.
Making Outbound Calls
To initiate outbound calls, refer to the Plivo WebSocket Integration guide for complete dial-out implementation details and examples.
Custom Data Requirements
If you need to pass custom parameters to your bot beyond the standard call information (CallUUID, caller details) that Plivo automatically provides, you can pass custom data through query parameters in the WebSocket URL.
For Pipecat Cloud deployment, custom data must be passed through the special body
parameter, which should be base64-encoded to comply with telephony vendor requirements:
Example XML server with custom data for Pipecat Cloud:
import base64
import json
@app.get("/plivo-xml")
async def plivo_xml(
agent: str = Query(...),
org: str = Query(...),
# Your custom parameters
user_id: str = Query(None),
session_type: str = Query("default"),
campaign_id: str = Query(None),
):
# Prepare custom data
custom_data = {
"user_id": user_id,
"session_type": session_type,
"campaign_id": campaign_id
}
# Base64 encode the data
encoded_data = base64.b64encode(json.dumps(custom_data).encode()).decode()
# Generate XML with encoded data, using & for &
xml = f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Stream bidirectional="true" keepCallAlive="true" contentType="audio/x-mulaw;rate=8000">
wss://api.pipecat.daily.co/ws/plivo?serviceHost={agent}.{org}&body={encoded_data}
</Stream>
</Response>"""
return Response(content=xml, media_type="application/xml")
Accessing custom data in your bot:
import base64
import json
from pipecat.runner.types import RunnerArguments
async def bot(args: RunnerArguments):
# Access the custom data from the body parameter
if args.body:
# Base64 decode the data
decoded_data = base64.b64decode(args.body).decode()
custom_data = json.loads(decoded_data)
user_id = custom_data.get("user_id")
session_type = custom_data.get("session_type")
campaign_id = custom_data.get("campaign_id")
# Use the custom data to configure your bot
# ...
This approach allows you to inject custom data while leveraging Pipecat Cloud’s managed bot infrastructure.
Advanced Call Control
Your bot can control the active call by leveraging Plivo’s Voice API with the CallUUID
that’s automatically provided to your bot. This enables capabilities such as:
- Recording conversations
- Playing audio prompts
- Gathering DTMF input (keypad presses)
- Ending calls programmatically
For examples of these advanced features, refer to the sample implementation in the pipecat-examples repo and Plivo’s Voice API documentation.
Production Deployment
For production use, deploy your XML server to a reliable hosting platform instead of using ngrok.
Update your Plivo phone number configuration to use your production server URL.