jyapayne/vocode-python
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

dangling pre open-source fix

Browse source
This commit is contained in:
Ajay Raj 2023-03-29 12:15:01 -07:00
commit 577a4e798e
3 changed files with 0 additions and 170 deletions
Download patch file Download diff file Expand all files Collapse all files

62
vocode/streaming/telephony/inbound_call_server.py
View file

@ -1,62 +0,0 @@
from fastapi import FastAPI, Response, Form
from typing import Optional
import requests
import uvicorn
import vocode
from vocode.streaming.models.transcriber import TranscriberConfig
from vocode.streaming.models.synthesizer import SynthesizerConfig
from vocode.streaming.models.agent import AgentConfig
from vocode.streaming.models.telephony import (
CreateInboundCall,
TwilioConfig,
TwilioConfig,
)
class InboundCallServer:
def __init__(
self,
agent_config: AgentConfig,
transcriber_config: Optional[TranscriberConfig] = None,
synthesizer_config: Optional[SynthesizerConfig] = None,
response_on_rate_limit: Optional[str] = None,
twilio_config: Optional[TwilioConfig] = None,
):
self.agent_config = agent_config
self.transcriber_config = transcriber_config
self.synthesizer_config = synthesizer_config
self.app = FastAPI()
self.app.post("/vocode")(self.handle_call)
self.response_on_rate_limit = (
response_on_rate_limit
or "The line is really busy right now, check back later!"
)
self.twilio_config = twilio_config
self.vocode_inbound_call_url = f"https://{vocode.base_url}/create_inbound_call"
async def handle_call(self, twilio_sid: str = Form(alias="CallSid")):
response = requests.post(
self.vocode_inbound_call_url,
headers={"Authorization": f"Bearer {vocode.api_key}"},
json=CreateInboundCall(
agent_config=self.agent_config,
twilio_sid=twilio_sid,
transcriber_config=self.transcriber_config,
synthesizer_config=self.synthesizer_config,
twilio_config=self.twilio_config,
).dict(),
)
if response.status_code == 429:
return Response(
f"<Response><Say>{self.response_on_rate_limit}</Say></Response>",
media_type="application/xml",
)
assert response.ok, response.text
return Response(
response.text,
media_type="application/xml",
)
def run(self, host="localhost", port=3000):
uvicorn.run(self.app, host=host, port=port)

40
vocode/streaming/telephony/inbound_call_user_agent_server.py
View file

@ -1,40 +0,0 @@
from typing import Optional, Union
from vocode.streaming.models.telephony import TwilioConfig
from vocode.streaming.telephony.inbound_call_server import InboundCallServer
from vocode.streaming.models.agent import (
RESTfulAgentEnd,
RESTfulAgentInput,
RESTfulAgentText,
RESTfulUserImplementedAgentConfig,
)
from vocode.streaming.models.transcriber import (
TranscriberConfig,
)
from vocode.streaming.models.synthesizer import SynthesizerConfig
class InboundCallUserAgentServer(InboundCallServer):
def __init__(
self,
agent_config: RESTfulUserImplementedAgentConfig,
transcriber_config: Optional[TranscriberConfig] = None,
synthesizer_config: Optional[SynthesizerConfig] = None,
response_on_rate_limit: Optional[str] = None,
twilio_config: Optional[TwilioConfig] = None,
):
super().__init__(
agent_config=agent_config,
transcriber_config=transcriber_config,
synthesizer_config=synthesizer_config,
response_on_rate_limit=response_on_rate_limit,
twilio_config=twilio_config,
)
assert isinstance(agent_config, RESTfulUserImplementedAgentConfig), "agent_config must be a RESTfulUserImplementedAgentConfig"
self.app.post("/respond")(self.respond_rest)
async def respond(self, human_input, conversation_id) -> Union[RESTfulAgentText, RESTfulAgentEnd]:
raise NotImplementedError
async def respond_rest(self, request: RESTfulAgentInput) -> Union[RESTfulAgentText, RESTfulAgentEnd]:
return await self.respond(request.human_input, request.conversation_id)

68
vocode/streaming/telephony/outbound_call.py
View file

@ -1,68 +0,0 @@
from typing import Optional
import requests
import vocode
from vocode.streaming.models.agent import AgentConfig
from vocode.streaming.models.synthesizer import SynthesizerConfig
from vocode.streaming.models.transcriber import TranscriberConfig
from ..models.telephony import (
CallEntity,
CreateOutboundCall,
EndOutboundCall,
TwilioConfig,
)
class OutboundCall:
def __init__(
self,
recipient: CallEntity,
caller: CallEntity,
agent_config: AgentConfig,
transcriber_config: Optional[TranscriberConfig] = None,
synthesizer_config: Optional[SynthesizerConfig] = None,
conversation_id: Optional[str] = None,
twilio_config: Optional[TwilioConfig] = None,
):
self.recipient = recipient
self.caller = caller
self.agent_config = agent_config
self.transcriber_config = transcriber_config
self.synthesizer_config = synthesizer_config
self.conversation_id = conversation_id
self.twilio_config = twilio_config
self.vocode_create_outbound_call_url = (
f"https://{vocode.base_url}/create_outbound_call"
)
self.vocode_end_outbound_call_url = (
f"https://{vocode.base_url}/end_outbound_call"
)
def start(self) -> str:
response = requests.post(
self.vocode_create_outbound_call_url,
headers={"Authorization": f"Bearer {vocode.api_key}"},
json=CreateOutboundCall(
recipient=self.recipient,
caller=self.caller,
agent_config=self.agent_config,
transcriber_config=self.transcriber_config,
synthesizer_config=self.synthesizer_config,
conversation_id=self.conversation_id,
twilio_config=self.twilio_config,
).dict(),
)
assert response.ok, response.text
data = response.json()
self.conversation_id = data["id"]
def end(self) -> str:
response = requests.post(
self.vocode_end_outbound_call_url,
headers={"Authorization": f"Bearer {vocode.api_key}"},
json=EndOutboundCall(
call_id=self.conversation_id,
twilio_config=self.twilio_config,
).dict(),
)
assert response.ok or response.status_code == 404, response.text