tools: cp_deploy_tarball waits for live card; new cp_refresh_agent
All checks were successful
build / build (push) Successful in 37s

cp_deploy_tarball now blocks after the from-tarball POST, polling the new
agent's /.well-known/agent-card until version matches the deployed version
(180s timeout, exponential backoff capped at 10s). On success it returns
live_skills with each skill's input_schema so the builder LLM can verify
the advertised shape matches the code it wrote — catches the schema/code
skew that caused downstream 400s on call_agent.

cp_refresh_agent(name) is a new tool that forces the CP to re-fetch the
live card via GET /v1/agents/{name}. Use when a pod was redeployed out
of band and the CP cache is stale; cp_deploy_tarball handles the in-band
case automatically.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
robert
2026-05-15 19:49:02 -03:00
parent df499cea51
commit 4cade1f3bb

View File

@@ -6,11 +6,13 @@
""" """
from __future__ import annotations from __future__ import annotations
import asyncio
import base64 import base64
import io import io
import json import json
import re import re
import tarfile import tarfile
import time
from dataclasses import dataclass from dataclasses import dataclass
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
@@ -31,6 +33,41 @@ class ToolContext:
cp_jwt: str | None = None cp_jwt: str | None = None
async def _wait_for_live_card(
url: str | None,
expected_version: str | None,
timeout_s: float = 180.0,
) -> tuple[bool, dict[str, Any]]:
"""Poll ``{url}/.well-known/agent-card`` until ``version`` matches
``expected_version`` or ``timeout_s`` elapses. Returns
``(matched, last_seen_card)``. ``last_seen_card`` is ``{}`` if the
endpoint never responded with a parseable card.
Backoff: 2s → 1.5x per try, capped at 10s. Tolerates pod-not-ready
(any HTTPError or 4xx/5xx) — just keeps polling.
"""
if not url or not expected_version:
return False, {}
deadline = time.monotonic() + timeout_s
last_card: dict[str, Any] = {}
interval = 2.0
while time.monotonic() < deadline:
try:
async with httpx.AsyncClient(timeout=5.0) as c:
r = await c.get(f"{url.rstrip('/')}/.well-known/agent-card")
if r.status_code == 200:
card = r.json() or {}
if isinstance(card, dict):
last_card = card
if card.get("version") == expected_version:
return True, card
except (httpx.HTTPError, ValueError):
pass
await asyncio.sleep(interval)
interval = min(interval * 1.5, 10.0)
return False, last_card
def _s3(ctx: ToolContext) -> Any: def _s3(ctx: ToolContext) -> Any:
return boto3.client( return boto3.client(
"s3", "s3",
@@ -180,7 +217,20 @@ def build_tools(ctx: ToolContext) -> list[Any]:
@tool @tool
async def cp_deploy_tarball(name: str, version: str = "0.1.0", public: bool = True) -> str: async def cp_deploy_tarball(name: str, version: str = "0.1.0", public: bool = True) -> str:
"""Tar the agent at ``agents/<name>/`` and POST it to the control """Tar the agent at ``agents/<name>/`` and POST it to the control
plane's ``/v1/agents/from-tarball`` endpoint on the user's behalf. plane's ``/v1/agents/from-tarball`` endpoint on the user's behalf,
then BLOCK until the live ``.well-known/agent-card`` reports the
new version (or 180s timeout).
Returns JSON with ``ok``, ``live``, ``url``, ``version``,
``live_version``, and ``live_skills[{name, input_schema}]``. When
``live=true`` the agent is callable; when ``live=false`` the
deploy POST succeeded but the pod never came up with the new
version — caller should treat this as a failure, not success.
Compare ``live_skills[].input_schema`` against the code you wrote
to confirm the advertised schema matches the params the function
actually accepts. Mismatches cause downstream ``agent 400`` from
``call_agent``.
Requires the orchestrator to have forwarded the user's CP JWT Requires the orchestrator to have forwarded the user's CP JWT
(the platform does this automatically when this agent's Card (the platform does this automatically when this agent's Card
@@ -222,14 +272,89 @@ def build_tools(ctx: ToolContext) -> list[Any]:
if r.status_code >= 400: if r.status_code >= 400:
return json.dumps({"error": f"cp {r.status_code}", "detail": r.text[:1000]}) return json.dumps({"error": f"cp {r.status_code}", "detail": r.text[:1000]})
body = r.json() body = r.json()
name_ = body.get("name")
version_ = body.get("version")
url_ = body.get("url")
# CP returns immediately with status="building". The agent isn't
# actually callable until the pod rolls + serves the new card.
# Poll /.well-known/agent-card until the live card reports the
# version we just deployed (or timeout). Surface the live skills
# so the LLM can verify the input_schema actually matches the
# code it wrote — catches the schema/code skew that causes
# downstream 400s on call_agent.
live, live_card = await _wait_for_live_card(url_, version_)
out: dict[str, Any] = {
"ok": live,
"name": name_,
"version": version_,
"status": body.get("status"),
"url": url_,
"live": live,
}
if live:
out["live_version"] = live_card.get("version")
out["live_skills"] = [
{
"name": s.get("name"),
"input_schema": s.get("input_schema") or {},
}
for s in (live_card.get("skills") or [])
]
else:
out["error"] = (
"deploy did not become live within timeout — pod may still"
" be rolling or build failed. Call cp_deploy_tarball again"
" or inspect via discover_agent before declaring done."
)
if live_card:
out["last_seen_version"] = live_card.get("version")
return json.dumps(out)
@tool
async def cp_refresh_agent(name: str) -> str:
"""Force the control plane to re-fetch the agent's live
``/.well-known/agent-card`` and update its stored copy.
Use this when the pod was redeployed out-of-band (CI bump, manual
rollout) and the CP's cached card is stale — symptoms are stale
``input_schema`` in ``list_my_agents`` / ``discover_agent`` and
4xx from ``call_agent`` on what should be a valid skill call.
``cp_deploy_tarball`` already does this automatically by polling
the live card, so you only need ``cp_refresh_agent`` when you
did NOT deploy through this tool.
Returns JSON: the fresh ``AgentDetailOut`` (with ``card`` +
``skills`` + ``input_schema``), or ``{error, ...}``.
"""
if not ctx.cp_jwt:
return json.dumps({ return json.dumps({
"ok": True, "name": body.get("name"), "version": body.get("version"), "error": "no CP JWT forwarded — agent declaration missing wants_cp_jwt=True",
"status": body.get("status"), "url": body.get("url"), })
try:
async with httpx.AsyncClient(timeout=15.0) as c:
r = await c.get(
f"{settings.cp_url}/v1/agents/{name}",
headers={"authorization": f"bearer {ctx.cp_jwt}"},
)
except httpx.HTTPError as exc:
return json.dumps({"error": f"cp unreachable: {exc}"})
if r.status_code >= 400:
return json.dumps({"error": f"cp {r.status_code}", "detail": r.text[:1000]})
body = r.json() or {}
return json.dumps({
"ok": True,
"name": body.get("name"),
"version": body.get("version"),
"status": body.get("status"),
"url": body.get("url"),
"card": body.get("card") or {},
}) })
return [ return [
list_agent_files, write_agent_file, read_agent_file, list_agent_files, write_agent_file, read_agent_file,
test_agent_in_sandbox, cp_deploy_tarball, test_agent_in_sandbox, cp_deploy_tarball, cp_refresh_agent,
] ]