Reject deploy error URLs in builder

agent.py

@@ -186,10 +186,11 @@ class AgentBuilder(A2AAgent[BuilderConfig, NoAuth]):
                         raw = raw["input"]
                     await ctx.emit_progress(f"→ {tname}({_one_line(raw)})")
                 elif kind == "on_tool_end":
-                    summary = _summarize(tname, data.get("output"))
+                    output = data.get("output")
+                    summary = _summarize(tname, output)
                     await ctx.emit_progress(f"  {tname} ← {summary}")
-                    # Capture URL from any tool output that mentions one.
-                    deployed_url = deployed_url or _find_url(data.get("output"))
+                    if tname == "cp_deploy_tarball":
+                        deployed_url = deployed_url or _find_deploy_url(output, name)
                 elif kind == "on_chain_end" and not event.get("parent_ids"):
                     out = data.get("output")
                     if isinstance(out, dict):
@@ -240,7 +241,7 @@ class AgentBuilder(A2AAgent[BuilderConfig, NoAuth]):
             content = msg.get("content") if isinstance(msg, dict) else getattr(msg, "content", None)
             if isinstance(content, str) and content.strip():
                 last_reply = content
-        deployed_url = deployed_url or _find_url(last_reply)
+        deployed_url = deployed_url or _find_agent_url(last_reply, name)
 
         if not deployed_url:
             return {
@@ -304,19 +305,37 @@ def _summarize(name: str, output: Any) -> str:
     return str(output)[:120]
 
 
-def _find_url(value: Any) -> str | None:
+def _find_deploy_url(value: Any, agent_name: str) -> str | None:
+    if isinstance(value, str):
+        try:
+            value = json.loads(value)
+        except (json.JSONDecodeError, ValueError):
+            return _find_agent_url(value, agent_name)
+    if not isinstance(value, dict):
+        return None
+    if value.get("error"):
+        return None
+    if value.get("ok") is False or value.get("live") is False:
+        return None
+    return _find_agent_url(value.get("url") or value.get("live_url"), agent_name)
+
+
+def _find_agent_url(value: Any, agent_name: str) -> str | None:
+    expected = f"https://{agent_name}.a2acloud.io"
     if value is None:
         return None
     if isinstance(value, dict):
         for k in ("url", "live_url"):
             v = value.get(k)
-            if isinstance(v, str) and v.startswith("http"):
+            if isinstance(v, str) and v.rstrip("/") == expected:
                 return v
         # JSON-stringified outputs from tools.
-        return _find_url(json.dumps(value))
+        return _find_agent_url(json.dumps(value), agent_name)
     if isinstance(value, str):
-        m = _URL_RE.search(value)
-        return m.group(0) if m else None
+        for m in _URL_RE.finditer(value):
+            candidate = m.group(0).rstrip("/")
+            if candidate == expected:
+                return candidate
     return None
 
 

tests/test_builder_prompt.py

@@ -70,6 +70,39 @@ class FakeGraph:
         }
 
 
+class FakePlanLimitGraph:
+    async def astream_events(self, *_args, **_kwargs):
+        yield {
+            "event": "on_tool_end",
+            "name": "cp_deploy_tarball",
+            "data": {
+                "output": {
+                    "error": "cp 402",
+                    "detail": (
+                        '{"detail":{"error":"plan_limit","feature":"agents",'
+                        '"upgrade_url":"https://app.a2acloud.io/billing"}}'
+                    ),
+                }
+            },
+        }
+        yield {
+            "event": "on_chain_end",
+            "parent_ids": [],
+            "data": {
+                "output": {
+                    "messages": [
+                        {
+                            "content": (
+                                "Deployment failed. Upgrade: "
+                                "https://app.a2acloud.io/billing"
+                            )
+                        }
+                    ]
+                }
+            },
+        }
+
+
 class AgentBuilderBuildTests(unittest.IsolatedAsyncioTestCase):
     async def test_build_captures_nested_graph_state_and_deployed_url(self) -> None:
         ctx = FakeRunContext()
@@ -87,6 +120,22 @@ class AgentBuilderBuildTests(unittest.IsolatedAsyncioTestCase):
         self.assertIn("Deployed at https://generated-agent.a2acloud.io", result["reply"])
         self.assertIn("deployed: https://generated-agent.a2acloud.io", ctx.progress)
 
+    async def test_build_does_not_treat_billing_url_as_deployed_agent(self) -> None:
+        ctx = FakeRunContext()
+
+        with patch("agent.build_agent_builder", return_value=FakePlanLimitGraph()):
+            result = await AgentBuilder().build(
+                ctx,
+                name="generated-agent",
+                prompt="make a test agent",
+                public=False,
+            )
+
+        self.assertEqual(result["ok"], False)
+        self.assertNotIn("url", result)
+        self.assertIn("no live URL was found", result["warning"])
+        self.assertNotIn("deployed: https://app.a2acloud.io/billing", ctx.progress)
+
 
 class BuilderPromptTests(unittest.TestCase):
     def test_outer_agent_uses_500_recursion_limit(self) -> None: