This project is inspired by Vocode
Telephony Server is a powerful bridge that connects telephony providers (Twilio, Vonage, Plivo, etc.) with real-time communication platforms (LiveKit, Jay.so, Pipecat, etc.). It enables seamless call routing, robust metrics collection, and observability features for enhanced telephony operations.
- Call Routing: Intelligent call forwarding and routing between providers and communication platforms.
- Multi-Provider Support: Seamless integration with Twilio, Vonage, Plivo, and more.
- Real-time Communication: Connects with services like LiveKit, Jay.so, and Pipecat.
- Metrics & Observability: Capture call analytics, monitor system health, and track call quality.
- Scalability: Built for high-performance and scalable architectures.
- Webhooks & Event Handling: Customizable event-driven architecture.
Ensure you have the following installed:
- Docker (Recommended for deployment)
- Twilio, Vonage, or Plivo Accounts
- LiveKit, Jay.so, or Pipecat Accounts
git clone https://github.com/steinathan/telephony-server.git
cd telephony-serverCreate a .env file and configure your credentials:
BASE_URL=6s6sgpwn-1338.uks1.devtunnels.ms
TWILIO_ACCOUNT_SID=
TWILIO_AUTH_TOKEN=
DEEPGRAM_API_KEY=
OPENAI_API_KEY=
ELEVENLABS_API_KEY=
TO_PHONE=
FROM_PHONE=uv sync --frozenMake an outbound phone call
import os
from dotenv import load_dotenv
from streaming_providers.models import BaseMessage
from streaming_providers.pipecat.pipecat import PipecatStreamingConfig
from telephony.config_manager.redis_config_manager import RedisConfigManager
from telephony.models.telephony import TwilioConfig
from telephony.outbound_call import OutboundCall
from telephony.utils.strings import create_conversation_id
load_dotenv()
BASE_URL = os.environ["BASE_URL"]
agent_name = "Navi"
system_prompt = """
You are {agent_name}, an AI-powered taxi dispatcher handling customer calls 24/7 with accuracy and professionalism. You assist customers in booking rides, managing driver availability, and resolving inquiries in a natural, efficient, and friendly manner.
You prioritize clear communication, confirming pickup details, estimating wait times, and ensuring a smooth booking experience. If a request cannot be fulfilled, offer alternative solutions or politely escalate to a human agent.
Maintain a warm, professional, and helpful tone. If a user is frustrated, reassure them and offer to help resolve their issue. Always confirm key details and minimize friction in the booking process.
---
You will be penalized if you don't follow the conversation stages below:
## Example Conversation Stages:
1. Greeting & Intent Detection
Customer: "Hi, I need a taxi."
{agent_name}: "Sure! Where would you like to be picked up from?"
Customer: "I want to check my booking."
{agent_name}: "Of course! Can you provide your booking reference or phone number?"
Customer: "How much to the airport?"
{agent_name}: "I can check that for you. What’s your pickup location?"
---
2. Booking a Ride
{agent_name}: "Where would you like to go?"
Customer: "To the airport."
{agent_name}: "Got it. What’s your pickup location?"
Customer: "123 Main Street."
{agent_name}: "Thanks! A taxi will arrive in 10 minutes. Would you like me to send a confirmation text?"
---
3️⃣ Handling Ride Inquiries
Customer: "Where’s my taxi?"
{agent_name}: "Let me check. Your driver is 3 minutes away in a white Toyota Corolla. Would you like me to share the driver’s contact details?"
Customer: "I need to cancel my ride."
{agent_name}: "I can help with that. Just to confirm, you’d like to cancel your ride from 123 Main Street to the airport?"
---
4️. Escalation to Human Agent
If the you cannot resolve the request (e.g., lost item, special assistance request, driver complaints), it should politely escalate:
{agent_name}: "I’ll transfer you to a live agent who can assist you further. Please hold for a moment."
""".format(agent_name=agent_name)
async def main():
config_manager = RedisConfigManager()
outbound_call = OutboundCall(
conversation_id=create_conversation_id("outbound"),
base_url=BASE_URL,
to_phone=os.environ["TO_PHONE"],
from_phone=os.environ["FROM_PHONE"],
config_manager=config_manager,
telephony_config=TwilioConfig(
account_sid=os.environ["TWILIO_ACCOUNT_SID"],
auth_token=os.environ["TWILIO_AUTH_TOKEN"],
),
telephony_params={
"Record": "false",
},
streaming_provider_config=PipecatStreamingConfig(
deepgram_api_key=os.environ["DEEPGRAM_API_KEY"],
openai_api_key=os.environ["OPENAI_API_KEY"],
elevenlabs_api_key=os.environ["ELEVENLABS_API_KEY"],
llm_model="gpt-4o-mini",
greeting_message=BaseMessage(
message="Thanks for calling, I'm Navi, your AI-powered taxi dispatcher. How can I help you today?"
),
prompt_premble=BaseMessage(message=system_prompt),
),
)
input("Press enter to start call...")
await outbound_call.start()
if __name__ == "__main__":
import asyncio
asyncio.run(main())You'd need to run the server before making the call, also make sure redis is working locally
$ uvicorn apps.telephony_app.main:app --port 1338Then make an outbound call
$ python -m apps.telephony_app.outbound_callYou can deploy using Docker:
docker-compose up --build -d- Prometheus & Grafana for call metrics visualization.
- Logging with ELK Stack (Elasticsearch, Logstash, Kibana).
We welcome contributions! Feel free to open issues and submit pull requests.
MIT License
Psst, I'm looking for a good job in the AI-ish, voice AI space, if you're hiring, kindly reach out to me on linkedin
Author: Navicstein Chinemerem
Contact: navicsteinrotciv@gmail.com
Linkedin:: https://www.linkedin.com/in/navicstein
Buy Me a Coffee: buymeacoffee.com/navicstein