Get started with this code:
from fastrtc import Stream, ReplyOnPause
def generate_audio(audio: tuple[int, np.ndarray]):
for chunk in generate_audio(audio):
yield (sample_rate, chunk)
stream = Stream(
modality="audio",
mode="send-receive",
handler=ReplyOnPause(response),
)
stream.fastphone()
This prints out the following:
"Call +1 877-713-4471 and input code 4582 when prompted"
"""
return html_content
@app.post("/register")
async def register(body: RegisterRequest, request: Request):
# Check Auth token valid with whoami
try:
whoami(request.headers["Authorization"])
except HTTPError:
return JSONResponse(status_code=401, content={"error": "Unauthorized"})
async with Lock():
codes = [s.code for s in sessions]
while (code := random.randint(0, 999999)) in codes:
pass
session = Session(url=body.url, code=code)
sessions.append(session)
return JSONResponse(
status_code=200,
content={"code": session.code, "phone": os.environ["TWILIO_PHONE_NUMBER"]},
)
@app.post("/incoming")
async def incoming(request: Request):
response = VoiceResponse()
gather = Gather(action="/connect", method="GET", timeout=10)
gather.say("Please enter your code,\nfollowed by the pound sign")
response.append(gather)
response.say("We didn't receive any input. Goodbye!")
return HTMLResponse(content=str(response), media_type="application/xml")
@app.get("/connect")
async def handle_incoming_call(Digits: str):
from twilio.twiml.voice_response import Connect, VoiceResponse
session = next((s for s in sessions if s.code == int(Digits)), None)
if session is None:
response = VoiceResponse()
response.say("Invalid code. Goodbye!")
return HTMLResponse(content=str(response), media_type="application/xml")
response = VoiceResponse()
response.say("Connecting to the AI assistant.")
connect = Connect()
connect.stream(url=f"wss://{session.url}/telephone/handler")
response.append(connect)
response.say("The call has been disconnected.")
return HTMLResponse(content=str(response), media_type="application/xml")
@app.post("/unregister")
async def unregister(body: UnregisterRequest, request: Request):
# Check Auth token valid with whoami
try:
whoami(request.headers["Authorization"])
except HTTPError:
return JSONResponse(status_code=401, content={"error": "Unauthorized"})
async with Lock():
index = next((i for i, s in enumerate(sessions) if s.url == body.url and s.code == body.code), None)
sessions.pop(index)
return JSONResponse(status_code=200, content={"message": "Unregistered"})
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860)