Poll private deployed agents by canonical URL

agent_builder/tools.py

@@ -42,6 +42,20 @@ class ToolContext:
     grant_token: str | None = None
 
 
+def _canonical_agent_url(name: str | None) -> str | None:
+    if not isinstance(name, str) or not name.strip():
+        return None
+    return f"https://{name.strip()}.a2acloud.io"
+
+
+def _deploy_poll_url(body: dict[str, Any], name: str | None) -> str | None:
+    for key in ("expected_url", "url"):
+        value = body.get(key)
+        if isinstance(value, str) and value.strip():
+            return value.strip()
+    return _canonical_agent_url(name)
+
+
 async def _wait_for_live_card(
     url: str | None,
     expected_version: str | None,
@@ -541,7 +555,7 @@ def build_tools(ctx: ToolContext) -> list[Any]:
         body = r.json()
         name_ = body.get("name")
         version_ = body.get("version")
-        url_ = body.get("url")
+        url_ = _deploy_poll_url(body, str(name_) if name_ is not None else name)
         head_sha = body.get("head_sha")
         if isinstance(head_sha, str) and head_sha:
             _write_builder_state(

tests/test_template_init.py

@@ -12,6 +12,7 @@ from agent_builder.tools import (
     _BUILDER_STATE_FILE,
     _BUILDER_INTERNAL_PREFIX,
     _parse_manifest_json,
+    _deploy_poll_url,
     _deployment_drift_error,
     _files_from_tarball,
     _parse_supporting_skill_files,
@@ -356,6 +357,22 @@ class TemplateInitTests(unittest.TestCase):
 
         self.assertIn("no CP JWT", result["error"])
 
+    def test_deploy_poll_url_falls_back_to_private_canonical_route(self) -> None:
+        self.assertEqual(
+            _deploy_poll_url({"url": None}, "private-agent"),
+            "https://private-agent.a2acloud.io",
+        )
+        self.assertEqual(
+            _deploy_poll_url(
+                {
+                    "expected_url": "https://expected.a2acloud.io",
+                    "url": "https://public.a2acloud.io",
+                },
+                "private-agent",
+            ),
+            "https://expected.a2acloud.io",
+        )
+
 
 def _tarball(files: dict[str, str]) -> bytes:
     buf = io.BytesIO()