feat: add configured base URL support to client

src/agora_agent/core/domain.py

@@ -1,9 +1,12 @@
 # This file was auto-generated by Fern from our API Definition.
 
 import asyncio
+import os
+import posixpath
 import socket
 import threading
 import time
+import urllib.parse
 from enum import IntEnum
 from typing import List, Optional
 
@@ -38,6 +41,27 @@ class Area(IntEnum):
 GLOBAL_API_PATH_SUFFIX = "/api/conversational-ai-agent"
 CN_API_PATH_SUFFIX = "/cn/api/conversational-ai-agent"
 
+_API_BASE_URL_ENV = "AGORA_AGENTS_API_BASE_URL"
+
+
+def _convo_ai_api_path_suffix(area: Area) -> str:
+    if area == Area.CN:
+        return CN_API_PATH_SUFFIX
+    return GLOBAL_API_PATH_SUFFIX
+
+
+def _get_configured_base_url(area: Area) -> Optional[str]:
+    raw_base_url = os.getenv(_API_BASE_URL_ENV, "").strip()
+    if not raw_base_url:
+        return None
+
+    try:
+        parsed_url = urllib.parse.urlsplit(raw_base_url)
+        path = posixpath.join(parsed_url.path, _convo_ai_api_path_suffix(area).lstrip("/"))
+        return parsed_url._replace(path=path).geturl()
+    except ValueError as exc:
+        raise ValueError(f"invalid {_API_BASE_URL_ENV}: {exc}") from exc
+
 
 class Domain:
     """Domain contains the regional prefixes and domain suffixes for an area"""
@@ -142,6 +166,7 @@ def __init__(self, domain_area: Area):
             raise ValueError("invalid domain area")
 
         self._domain_area = domain_area
+        self._configured_base_url = _get_configured_base_url(domain_area)
         self._domain_suffixes = list(domain_config.major_domain_suffixes)
         self._region_prefixes = list(domain_config.region_domain_prefixes)
         self._current_region_prefixes = list(self._region_prefixes)
@@ -156,6 +181,9 @@ def _domain_need_update(self) -> bool:
 
     def select_best_domain(self) -> None:
         """SelectBestDomain uses DNS resolution to select the best available domain (sync)"""
+        if self._configured_base_url is not None:
+            return
+
         if not self._domain_need_update():
             return
 
@@ -166,6 +194,9 @@ def select_best_domain(self) -> None:
 
     async def select_best_domain_async(self) -> None:
         """SelectBestDomain uses DNS resolution to select the best available domain (async)"""
+        if self._configured_base_url is not None:
+            return
+
         if not self._domain_need_update():
             return
 
@@ -178,6 +209,9 @@ async def select_best_domain_async(self) -> None:
 
     def next_region(self) -> None:
         """NextRegion cycles to the next region prefix in the pool"""
+        if self._configured_base_url is not None:
+            return
+
         with self._lock:
             self._current_region_prefixes = self._current_region_prefixes[1:]
             if len(self._current_region_prefixes) == 0:
@@ -191,10 +225,12 @@ def _select_domain(self, domain: str) -> None:
     def get_current_url(self) -> str:
         """GetCurrentURL returns the current URL based on the selected region and domain"""
         with self._lock:
+            if self._configured_base_url is not None:
+                return self._configured_base_url
+
             current_region = self._current_region_prefixes[0]
             current_domain = self._current_domain
-            path_suffix = CN_API_PATH_SUFFIX if self._domain_area == Area.CN else GLOBAL_API_PATH_SUFFIX
-            return f"https://{current_region}.{current_domain}{path_suffix}"
+            return f"https://{current_region}.{current_domain}{_convo_ai_api_path_suffix(self._domain_area)}"
 
     def get_area(self) -> Area:
         """Get the current area"""

tests/custom/test_domain.py

@@ -0,0 +1,115 @@
+from typing import List
+
+import httpx
+import pytest
+
+from agora_agent import Agora
+from agora_agent.core.domain import Area, AsyncResolverImpl, Pool, ResolverImpl
+
+_API_BASE_URL_ENV = "AGORA_AGENTS_API_BASE_URL"
+
+
+@pytest.mark.parametrize(
+    ("area", "expected"),
+    [
+        (Area.US, "https://api-test.agora.io/api/conversational-ai-agent"),
+        (Area.EU, "https://api-test.agora.io/api/conversational-ai-agent"),
+        (Area.AP, "https://api-test.agora.io/api/conversational-ai-agent"),
+        (Area.CN, "https://api-test.agora.io/cn/api/conversational-ai-agent"),
+    ],
+)
+def test_pool_uses_configured_base_url(
+    monkeypatch: pytest.MonkeyPatch,
+    area: Area,
+    expected: str,
+) -> None:
+    monkeypatch.setenv(_API_BASE_URL_ENV, "https://api-test.agora.io/")
+
+    assert Pool(area).get_current_url() == expected
+
+
+@pytest.mark.parametrize(
+    ("base_url", "expected"),
+    [
+        (
+            "https://staging.example.com/",
+            "https://staging.example.com/api/conversational-ai-agent",
+        ),
+        (
+            "http://localhost:8080",
+            "http://localhost:8080/api/conversational-ai-agent",
+        ),
+        (
+            "https://user:password@staging.example.com/gateway?debug=true#section",
+            "https://user:password@staging.example.com/gateway/api/conversational-ai-agent?debug=true#section",
+        ),
+    ],
+)
+def test_pool_accepts_configured_base_url(
+    monkeypatch: pytest.MonkeyPatch,
+    base_url: str,
+    expected: str,
+) -> None:
+    monkeypatch.setenv(_API_BASE_URL_ENV, base_url)
+
+    assert Pool(Area.US).get_current_url() == expected
+
+
+def test_configured_base_url_disables_dynamic_routing(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv(_API_BASE_URL_ENV, "https://api-test.agora.io")
+    pool = Pool(Area.US)
+    pool._resolver = _FailingResolver()
+
+    expected = "https://api-test.agora.io/api/conversational-ai-agent"
+    for _ in range(3):
+        pool.next_region()
+        pool.select_best_domain()
+        assert pool.get_current_url() == expected
+
+
+@pytest.mark.asyncio
+async def test_configured_base_url_disables_async_domain_selection(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv(_API_BASE_URL_ENV, "https://api-test.agora.io")
+    pool = Pool(Area.CN)
+    pool._async_resolver = _FailingAsyncResolver()
+
+    await pool.select_best_domain_async()
+
+    assert pool.get_current_url() == "https://api-test.agora.io/cn/api/conversational-ai-agent"
+
+
+def test_pool_without_configured_base_url_keeps_regional_routing(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv(_API_BASE_URL_ENV, raising=False)
+    pool = Pool(Area.US)
+
+    assert pool.get_current_url() == "https://api-us-west-1.agora.io/api/conversational-ai-agent"
+
+    pool.next_region()
+
+    assert pool.get_current_url() == "https://api-us-east-1.agora.io/api/conversational-ai-agent"
+
+
+def test_client_uses_configured_base_url(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv(_API_BASE_URL_ENV, "https://api-test.agora.io")
+
+    with httpx.Client() as httpx_client:
+        client = Agora(
+            area=Area.CN,
+            app_id="0" * 32,
+            app_certificate="1" * 32,
+            httpx_client=httpx_client,
+        )
+
+        expected = "https://api-test.agora.io/cn/api/conversational-ai-agent"
+        assert client.get_current_url() == expected
+        assert client._client_wrapper.get_base_url() == expected
+
+
+class _FailingResolver(ResolverImpl):
+    def resolve(self, domains: List[str], region_prefix: str) -> str:
+        raise AssertionError("resolver should not be called")
+
+
+class _FailingAsyncResolver(AsyncResolverImpl):
+    async def resolve_async(self, domains: List[str], region_prefix: str) -> str:
+        raise AssertionError("async resolver should not be called")