recording

Author: aboutphilippe

agent_telephony/twilio_livekit/agent_twilio/.env.Example

@@ -1,18 +1,22 @@
 
 RESTACK_API_KEY=
+OPENAI_API_KEY=
+
 LIVEKIT_API_KEY=
 LIVEKIT_API_SECRET=
 LIVEKIT_URL=
 
-LIVEKIT_SIP_ADDRESS=
 TWILIO_PHONE_NUMBER=
 TWILIO_TRUNK_AUTH_USERNAME=
 TWILIO_TRUNK_AUTH_PASSWORD=
+TWILIO_TRUNK_TERMINATION_SIP_URL=
 
 ELEVEN_API_KEY=
 DEEPGRAM_API_KEY=
 OPENAI_API_KEY=
 
+GCP_CREDENTIALS=
+
 # Restack Cloud (Optional)
 
 # RESTACK_ENGINE_ID=<your-engine-id>

agent_telephony/twilio_livekit/agent_twilio/event_agent.py

@@ -11,14 +11,23 @@ async def main(agent_id: str, run_id: str) -> None:
         agent_id=agent_id,
         run_id=run_id,
         event_name="call",
-        event_input={"messages": [{"role": "user", "content": "Tell me another joke"}]},
+        event_input={
+            "messages": [
+                {
+                    "role": "user",
+                    "content": "What is Restack framework?",
+                }
+            ]
+        },
     )
 
     sys.exit(0)
 
 
 def run_event_workflow() -> None:
-    asyncio.run(main(agent_id="your-agent-id", run_id="your-run-id"))
+    asyncio.run(
+        main(agent_id="agent-id", run_id="run-id")
+    )
 
 
 if __name__ == "__main__":

agent_telephony/twilio_livekit/agent_twilio/schedule_agent.py

@@ -14,7 +14,9 @@ async def main() -> None:
         agent_name=AgentStream.__name__, agent_id=agent_id
     )
 
-    await client.get_agent_result(agent_id=agent_id, run_id=run_id)
+    await client.get_agent_result(
+        agent_id=agent_id, run_id=run_id
+    )
 
     sys.exit(0)
 

agent_telephony/twilio_livekit/agent_twilio/src/agents/agent.py

@@ -37,6 +37,11 @@
         SendDataResponse,
         livekit_send_data,
     )
+    from src.functions.livekit_start_recording import (
+        EgressInfo,
+        LivekitStartRecordingInput,
+        livekit_start_recording,
+    )
     from src.functions.livekit_token import (
         LivekitTokenInput,
         livekit_token,
@@ -69,6 +74,13 @@ class PipelineMetricsEvent(BaseModel):
     latencies: str
 
 
+class AgentTwilioOutput(BaseModel):
+    recording_url: str
+    livekit_room_id: str
+    messages: list[Message]
+    context: str
+
+
 @agent.defn()
 class AgentTwilio:
     def __init__(self) -> None:
@@ -78,7 +90,9 @@ def __init__(self) -> None:
         self.room_id = ""
 
     @agent.event
-    async def messages(self, messages_event: MessagesEvent) -> list[Message]:
+    async def messages(
+        self, messages_event: MessagesEvent
+    ) -> list[Message]:
         log.info(f"Received message: {messages_event.messages}")
         self.messages.extend(messages_event.messages)
 
@@ -108,7 +122,9 @@ async def messages(self, messages_event: MessagesEvent) -> list[Message]:
                 ),
             )
 
-            self.messages.append(Message(role="assistant", content=fast_response))
+            self.messages.append(
+                Message(role="assistant", content=fast_response)
+            )
             return self.messages
         except Exception as e:
             error_message = f"Error during messages: {e}"
@@ -122,7 +138,9 @@ async def call(self, call_input: CallInput) -> None:
         agent_id = agent_info().workflow_id
         run_id = agent_info().run_id
         try:
-            sip_trunk_id = await agent.step(function=livekit_outbound_trunk)
+            sip_trunk_id = await agent.step(
+                function=livekit_outbound_trunk
+            )
             await agent.step(
                 function=livekit_call,
                 function_input=LivekitCallInput(
@@ -135,14 +153,17 @@ async def call(self, call_input: CallInput) -> None:
                 ),
             )
         except Exception as e:
-            error_message = f"Error during livekit_outbound_trunk: {e}"
+            error_message = (
+                f"Error during livekit_outbound_trunk: {e}"
+            )
             raise NonRetryableError(error_message) from e
 
     @agent.event
     async def say(self, say: str) -> SendDataResponse:
         log.info("Received say")
         return await agent.step(
-            function=livekit_send_data, function_input=LivekitSendDataInput(text=say)
+            function=livekit_send_data,
+            function_input=LivekitSendDataInput(text=say),
         )
 
     @agent.event
@@ -151,10 +172,11 @@ async def end(self, end: EndEvent) -> EndEvent:
         await agent.step(
             function=livekit_send_data,
             function_input=LivekitSendDataInput(
-                room_id=self.room_id, text="Thank you for calling restack. Goodbye!"
+                room_id=self.room_id,
+                text="Thank you for calling restack. Goodbye!",
             ),
         )
-        await agent.sleep(1)
+        await agent.sleep(3)
         await agent.step(function=livekit_delete_room)
 
         self.end = True
@@ -170,7 +192,10 @@ async def context(self, context: ContextEvent) -> str:
     async def pipeline_metrics(
         self, pipeline_metrics: PipelineMetricsEvent
     ) -> PipelineMetricsEvent:
-        log.info("Received pipeline metrics", pipeline_metrics=pipeline_metrics)
+        log.info(
+            "Received pipeline metrics",
+            pipeline_metrics=pipeline_metrics,
+        )
         return pipeline_metrics
 
     @agent.run
@@ -180,14 +205,34 @@ async def run(self) -> None:
             self.room_id = room.name
             await agent.step(
                 function=livekit_token,
-                function_input=LivekitTokenInput(room_id=self.room_id),
+                function_input=LivekitTokenInput(
+                    room_id=self.room_id
+                ),
+            )
+            recording: EgressInfo = await agent.step(
+                function=livekit_start_recording,
+                function_input=LivekitStartRecordingInput(
+                    room_id=self.room_id
+                ),
             )
             await agent.step(
                 function=livekit_dispatch,
-                function_input=LivekitDispatchInput(room_id=self.room_id),
+                function_input=LivekitDispatchInput(
+                    room_id=self.room_id
+                ),
             )
+
         except Exception as e:
             error_message = f"Error during agent run: {e}"
             raise NonRetryableError(error_message) from e
         else:
             await agent.condition(lambda: self.end)
+
+            recording_url = f"https://storage.googleapis.com/{recording.room_composite.file_outputs[0].gcp.bucket}/{recording.room_composite.file_outputs[0].filepath}"
+
+            return AgentTwilioOutput(
+                recording_url=recording_url,
+                livekit_room_id=self.room_id,
+                messages=self.messages,
+                context=self.context,
+            )

agent_telephony/twilio_livekit/agent_twilio/src/client.py

@@ -14,6 +14,9 @@
 api_address = os.getenv("RESTACK_ENGINE_API_ADDRESS")
 
 connection_options = CloudConnectionOptions(
-    engine_id=engine_id, address=address, api_key=api_key, api_address=api_address
+    engine_id=engine_id,
+    address=address,
+    api_key=api_key,
+    api_address=api_address,
 )
 client = Restack(connection_options)

agent_telephony/twilio_livekit/agent_twilio/src/functions/context_docs.py

@@ -3,10 +3,15 @@
 
 
 async def fetch_content_from_url(url: str) -> str:
-    async with aiohttp.ClientSession() as session, session.get(url) as response:
+    async with (
+        aiohttp.ClientSession() as session,
+        session.get(url) as response,
+    ):
         if response.status == 200:
             return await response.text()
-        error_message = f"Failed to fetch content: {response.status}"
+        error_message = (
+            f"Failed to fetch content: {response.status}"
+        )
         raise NonRetryableError(error_message)
 
 
@@ -16,7 +21,9 @@ async def context_docs() -> str:
         docs_content = await fetch_content_from_url(
             "https://docs.restack.io/llms-full.txt"
         )
-        log.info("Fetched content from URL", content=len(docs_content))
+        log.info(
+            "Fetched content from URL", content=len(docs_content)
+        )
 
         return docs_content
 

agent_telephony/twilio_livekit/agent_twilio/src/functions/livekit_call.py

@@ -19,7 +19,9 @@ class LivekitCallInput:
 
 
 @function.defn()
-async def livekit_call(function_input: LivekitCallInput) -> SIPParticipantInfo:
+async def livekit_call(
+    function_input: LivekitCallInput,
+) -> SIPParticipantInfo:
     try:
         livekit_api = api.LiveKitAPI()
 
@@ -32,13 +34,21 @@ async def livekit_call(function_input: LivekitCallInput) -> SIPParticipantInfo:
             play_dialtone=True,
         )
 
-        log.info("livekit_call CreateSIPParticipantRequest: ", request=request)
+        log.info(
+            "livekit_call CreateSIPParticipantRequest: ",
+            request=request,
+        )
 
-        participant = await livekit_api.sip.create_sip_participant(request)
+        participant = (
+            await livekit_api.sip.create_sip_participant(request)
+        )
 
         await livekit_api.aclose()
 
-        log.info("livekit_call SIPParticipantInfo:", participant=participant)
+        log.info(
+            "livekit_call SIPParticipantInfo:",
+            participant=participant,
+        )
     except Exception as e:
         error_message = f"livekit_call function failed: {e}"
         raise NonRetryableError(error_message) from e

agent_telephony/twilio_livekit/agent_twilio/src/functions/livekit_create_room.py

@@ -31,7 +31,9 @@ async def livekit_create_room() -> Room:
         await lkapi.aclose()
 
     except Exception as e:
-        error_message = f"livekit_create_room function failed: {e}"
+        error_message = (
+            f"livekit_create_room function failed: {e}"
+        )
         raise NonRetryableError(error_message) from e
 
     else:

agent_telephony/twilio_livekit/agent_twilio/src/functions/livekit_delete_room.py

@@ -20,12 +20,16 @@ async def livekit_delete_room() -> DeleteRoomResponse:
 
         run_id = function_info().workflow_run_id
 
-        deleted_room = await lkapi.room.delete_room(DeleteRoomRequest(room=run_id))
+        deleted_room = await lkapi.room.delete_room(
+            DeleteRoomRequest(room=run_id)
+        )
 
         await lkapi.aclose()
 
     except Exception as e:
-        error_message = f"livekit_delete_room function failed: {e}"
+        error_message = (
+            f"livekit_delete_room function failed: {e}"
+        )
         raise NonRetryableError(error_message) from e
 
     else:

agent_telephony/twilio_livekit/agent_twilio/src/functions/livekit_dispatch.py

@@ -16,7 +16,9 @@ class LivekitDispatchInput:
 
 
 @function.defn()
-async def livekit_dispatch(function_input: LivekitDispatchInput) -> AgentDispatch:
+async def livekit_dispatch(
+    function_input: LivekitDispatchInput,
+) -> AgentDispatch:
     try:
         lkapi = api.LiveKitAPI(
             url=os.getenv("LIVEKIT_API_URL"),
@@ -28,13 +30,19 @@ async def livekit_dispatch(function_input: LivekitDispatchInput) -> AgentDispatc
         agent_id = function_info().workflow_id
         run_id = function_info().workflow_run_id
 
-        metadata = {"agent_name": agent_name, "agent_id": agent_id, "run_id": run_id}
+        metadata = {
+            "agent_name": agent_name,
+            "agent_id": agent_id,
+            "run_id": run_id,
+        }
 
         room = function_input.room_id or run_id
 
         dispatch = await lkapi.agent_dispatch.create_dispatch(
             api.CreateAgentDispatchRequest(
-                agent_name=agent_name, room=room, metadata=str(metadata)
+                agent_name=agent_name,
+                room=room,
+                metadata=str(metadata),
             )
         )