adding support for the external call via proxy

Author: priyanshi-2003

.env.example

@@ -108,6 +108,11 @@ AUTOMATIC_OPENAI_STT_PROMPT="Transcribe Indian languages accurately. Recognize a
 AUTOMATIC_WRITE_ACTIONS_AUTHORIZED_USERS=user1@mail.com,user2@mail.com,...
 AUTOMATIC_ACTIONS_REQUIRE_AUTH=create_euler_offer,delete_euler_offer,...
 
+# Proxy Configuration
+AWS_PROXY_HOST=localhost
+AWS_PROXY_PORT=80
+CLOUD_PROVIDER=AWS
+
 # LangFuse Configuration
 ENABLE_LANGFUSE_PROMPTS=false
 LANGFUSE_SECRET_KEY=''

app/agents/voice/automatic/services/mcp/automatic_client.py

@@ -7,6 +7,7 @@
 
 from app.core.config import MCP_CLIENT_TIMEOUT
 from app.core.logger import logger
+from app.core.transport.http_client import create_http_client
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
 from pipecat.adapters.schemas.function_schema import FunctionSchema
 from app.agents.voice.automatic.types.models import (
@@ -33,7 +34,7 @@ def __init__(self, server_url: str, auth_token: str, context: Dict[str, Any]):
         self._server_url = server_url.strip()
         self._auth_token = auth_token
         self._context_b64 = base64.b64encode(json.dumps(context).encode()).decode()
-        self._client = httpx.AsyncClient(timeout=MCP_CLIENT_TIMEOUT)
+        self._client = create_http_client(timeout=MCP_CLIENT_TIMEOUT)
         self._demo_mode = context.get("enableDemoMode", False)
 
     async def post(

app/agents/voice/automatic/tools/breeze/analytics.py

@@ -9,6 +9,7 @@
 from pipecat.services.llm_service import FunctionCallParams
 from pipecat.adapters.schemas.function_schema import FunctionSchema
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from app.core.transport.http_client import create_http_client
 
 # These will be set by the initializer
 breeze_token: str | None = None
@@ -98,7 +99,7 @@ async def _make_breeze_request(params: FunctionCallParams, operational_tab: str)
     )
 
     try:
-        async with httpx.AsyncClient(timeout=30.0) as client:
+        async with create_http_client(timeout=30.0) as client:
             response = await client.post(api_url, json=payload, headers=headers)
             response.raise_for_status()  # Raise an exception for bad status codes
             response_json = response.json()
@@ -223,7 +224,7 @@ async def get_breeze_marketing_data(params: FunctionCallParams):
     logger.info(f"Requesting Breeze marketing data with payload: {json.dumps(payload)}")
 
     try:
-        async with httpx.AsyncClient() as client:
+        async with create_http_client() as client:
             response = await client.post(api_url, json=payload, headers=headers)
             response.raise_for_status()
             response_json = response.json()
@@ -326,7 +327,7 @@ async def get_breeze_address_data(params: FunctionCallParams):
     logger.info(f"Requesting Breeze address data with payload: {json.dumps(payload)}")
 
     try:
-        async with httpx.AsyncClient() as client:
+        async with create_http_client() as client:
             response = await client.post(api_url, json=payload, headers=headers)
             response.raise_for_status()
             response_json = response.json()

app/agents/voice/automatic/tools/breeze/utils.py

@@ -13,6 +13,7 @@
 
 from app.core.logger import logger
 from app.core.config import LIGHTHOUSE_APP_URL
+from app.core.transport.http_client import create_http_client
 
 
 def safe_construct_url(url: str) -> Optional[urlparse]:
@@ -115,7 +116,7 @@ async def get_current_shop_config_data(shop_url: str) -> Dict[str, Any]:
 
     try:
         logger.info(f"Fetching shop config from: {url}")
-        async with httpx.AsyncClient(timeout=30.0) as client:
+        async with create_http_client(timeout=30.0) as client:
             response = await client.get(url)
             response.raise_for_status()
             config_data = response.json()
@@ -165,7 +166,7 @@ async def patch_shop_config(
         "x-auth-token": breeze_token,
     }
     try:
-        async with httpx.AsyncClient(timeout=timeout) as client:
+        async with create_http_client(timeout=timeout) as client:
             response = await client.patch(url, headers=headers, json=config_data)
             response.raise_for_status()
             return response.json()

app/agents/voice/automatic/tools/juspay/analytics.py

@@ -14,6 +14,7 @@
     ApiSuccess,
     GeniusApiResponse,
 )
+from app.core.transport.http_client import create_http_client
 
 # This token will be set when the tools are initialized
 euler_token: str | None = None
@@ -115,7 +116,7 @@ async def _make_genius_api_request(
     )
 
     try:
-        async with httpx.AsyncClient(timeout=10.0) as client:
+        async with create_http_client(timeout=10.0) as client:
             response = await client.post(
                 GENIUS_API_URL, json=full_payload, headers=headers
             )
@@ -344,7 +345,7 @@ async def list_offers_by_filter(params: FunctionCallParams):
             f"Requesting Euler offers list from: {endpoint} | Payload: {json.dumps(payload, indent=2)}"
         )
 
-        async with httpx.AsyncClient(timeout=30.0) as client:
+        async with create_http_client(timeout=30.0) as client:
             response = await client.post(endpoint, json=payload, headers=headers)
 
             if response.status_code != 200:
@@ -736,7 +737,7 @@ async def create_euler_offer(params: FunctionCallParams):
             f"Making offer creation request to: {endpoint} | Payload: {json.dumps(api_payload, indent=2)}"
         )
 
-        async with httpx.AsyncClient(timeout=10.0) as client:
+        async with create_http_client(timeout=10.0) as client:
             response = await client.post(endpoint, json=api_payload, headers=headers)
 
             if response.status_code == 200:
@@ -890,7 +891,7 @@ async def find_offer_by_code(offer_code: str) -> dict | None:
             f"Searching for offer with code '{offer_code}' using endpoint: {endpoint}"
         )
 
-        async with httpx.AsyncClient(timeout=30.0) as client:
+        async with create_http_client(timeout=30.0) as client:
             response = await client.post(endpoint, json=search_payload, headers=headers)
 
             if response.status_code == 200:
@@ -983,7 +984,7 @@ async def delete_euler_offer(params: FunctionCallParams):
             f"Deleting offer with ID '{offer_id}' using endpoint: {delete_endpoint}"
         )
 
-        async with httpx.AsyncClient(timeout=30.0) as client:
+        async with create_http_client(timeout=30.0) as client:
             response = await client.post(
                 delete_endpoint, json=delete_payload, headers=headers
             )
@@ -1194,7 +1195,7 @@ async def update_euler_offer(params: FunctionCallParams):
 
             logger.info(f"Using status-only endpoint: {status_endpoint}")
 
-            async with httpx.AsyncClient(timeout=30.0) as client:
+            async with create_http_client(timeout=30.0) as client:
                 response = await client.post(
                     status_endpoint, json=status_payload, headers=headers
                 )
@@ -1544,7 +1545,7 @@ async def update_euler_offer(params: FunctionCallParams):
         )
         logger.info(f"Payload: {json.dumps(api_payload, indent=2)}")
 
-        async with httpx.AsyncClient(timeout=30.0) as client:
+        async with create_http_client(timeout=30.0) as client:
             response = await client.put(
                 update_endpoint, json=api_payload, headers=headers
             )

app/agents/voice/breeze_buddy/managers/calls.py

@@ -5,7 +5,7 @@
 from datetime import datetime, timezone, timedelta
 from app.core.logger import logger
 import uuid
-import aiohttp
+from app.core.transport.http_client import create_aiohttp_session
 from app.database.accessor import (
     get_leads_based_on_status_and_next_attempt,
     get_call_execution_config_by_merchant_id,
@@ -33,7 +33,7 @@ async def process_backlog_leads():
     Processes backlog leads and initiates calls.
     """
     logger.info("Processing backlog leads...")
-    async with aiohttp.ClientSession() as session:
+    async with create_aiohttp_session() as session:
         try:
             leads = await get_leads_based_on_status_and_next_attempt(
                 LeadCallStatus.BACKLOG, datetime.now(timezone.utc)

app/agents/voice/breeze_buddy/services/telephony/exotel/exotel.py

@@ -3,6 +3,7 @@
 import requests
 
 from app.core.logger import logger
+from app.core.transport.http_client import create_aiohttp_session
 
 from pipecat.serializers.exotel import ExotelFrameSerializer
 
@@ -52,7 +53,11 @@ def make_call(self, customer_mobile_number: str, outbound_number: str):
         logger.info(f"Payload: {payload}")
 
         try:
-            resp = requests.post(url, data=payload)
+            # Use centralized proxy configuration
+            proxy_url = get_proxy_config()
+            proxies = {"https": proxy_url, "http": proxy_url} if proxy_url else None
+
+            resp = requests.post(url, data=payload, proxies=proxies)
             logger.info(f"Exotel API response status: {resp.status_code}")
             logger.info(f"Exotel API response headers: {dict(resp.headers)}")
             logger.info(f"Exotel API response content: {resp.text}")

app/agents/voice/breeze_buddy/services/telephony/twilio/twilio.py

@@ -1,11 +1,13 @@
 from fastapi import WebSocket, HTTPException
 from twilio.rest import Client
+from twilio.http.http_client import TwilioHttpClient
 from twilio.twiml.voice_response import VoiceResponse, Connect, Stream
 
 from app.agents.voice.breeze_buddy.services.telephony.base_provider import (
     VoiceCallProvider,
 )
 from app.core import config
+from app.core.transport.http_client import create_aiohttp_session
 from app.agents.voice.breeze_buddy.workflows.order_confirmation.websocket_bot import (
     main as telephony_websocket_conn,
 )
@@ -22,9 +24,29 @@ async def _hang_up_call(self):
 
     def __init__(self, aiohttp_session):
         super().__init__(config, aiohttp_session)
-        self.client = Client(
-            self.config.TWILIO_ACCOUNT_SID, self.config.TWILIO_AUTH_TOKEN
-        )
+
+        # Create Twilio client with proper proxy configuration
+        self.client = self._create_twilio_client()
+
+    def _create_twilio_client(self) -> Client:
+        """Create Twilio client with proper proxy configuration using TwilioHttpClient"""
+        proxy_url = get_proxy_config()
+        account_sid = self.config.TWILIO_ACCOUNT_SID
+        auth_token = self.config.TWILIO_AUTH_TOKEN
+
+        if proxy_url:
+            logger.info(f"Configuring Twilio client with proxy: {proxy_url}")
+            # Use TwilioHttpClient with proxy configuration
+            proxy_client = TwilioHttpClient(
+                proxy={
+                    "http": proxy_url,
+                    "https": proxy_url,
+                }
+            )
+            return Client(account_sid, auth_token, http_client=proxy_client)
+        else:
+            logger.info("Creating Twilio client without proxy")
+            return Client(account_sid, auth_token)
 
     def hangup_call(self, call_sid: str):
         self.client.calls(call_sid).update(status="completed")

app/api/routers/breeze_buddy.py

@@ -23,7 +23,7 @@
 from app.agents.voice.breeze_buddy.workflows.order_confirmation.types import (
     BreezeOrderData,
 )
-import aiohttp
+from app.core.transport.http_client import create_aiohttp_session
 from app.agents.voice.breeze_buddy.managers.calls import (
     process_backlog_leads,
     handle_call_completion,
@@ -286,7 +286,7 @@ async def telephony_websocket_handler(
 
     logger.info(f"Handling websocket for {workflow}")
 
-    async with aiohttp.ClientSession() as session:
+    async with create_aiohttp_session() as session:
         try:
             provider = get_voice_provider(service_provider.upper(), session)
             provider.set_completion_callback(handle_call_completion)

app/core/config.py

@@ -281,6 +281,10 @@ def get_required_env(var_name: str) -> str:
 EXOTEL_SUBDOMAIN = os.getenv("EXOTEL_SUBDOMAIN", "api.exotel.com")
 EXOTEL_APPLET_APP_ID = os.getenv("EXOTEL_APPLET_APP_ID", "1044183")
 
+# Proxy Configuration
+AWS_PROXY_HOST = os.environ.get("AWS_PROXY_HOST")
+AWS_PROXY_PORT = os.environ.get("AWS_PROXY_PORT")
+CLOUD_ENVIRONMENT = os.environ.get("CLOUD_ENVIRONMENT", "GCP")  # AWS, GCP, AZURE, etc.
 
 # LangFuse Configuration
 ENABLE_LANGFUSE_PROMPTS = (