initial: deepagents-driven agent generator + deployer

agent.py

@@ -0,0 +1,219 @@
+"""agent-builder — generate + deploy new agents from natural language.
+
+Inner loop is a deepagents graph (see agent_builder/builder.py). The
+outer A2AAgent surface is a single skill ``build`` that takes a
+description, writes a scaffolded project to the user's workspace at
+``agents/<name>/``, tests it in a sandbox, deploys it through the
+control plane, and reports back the live URL.
+
+This agent only works when invoked through the platform orchestrator,
+because it needs three things forwarded from the caller:
+
+  1. ``ctx.workspace.bucket`` — the user's MinIO bucket
+  2. ``ctx.llm`` — caller's LLM creds (this is an expensive skill;
+     the user pays for inference directly via their own provider)
+  3. ``ctx.cp_jwt`` — the user's CP JWT, so cp_deploy_tarball can
+     POST to /v1/agents/from-tarball as the user.
+"""
+from __future__ import annotations
+
+import json
+import re
+from typing import Any
+
+from pydantic import BaseModel
+
+from a2a_pack import (
+    A2AAgent, LLMProvisioning, NoAuth, Pricing, RunContext, skill,
+)
+
+from agent_builder import BuilderContext, build_agent_builder
+
+
+class BuilderConfig(BaseModel):
+    pass
+
+
+_URL_RE = re.compile(r"https?://[a-zA-Z0-9._-]+\.[a-zA-Z]{2,}[\w/?#%&=.-]*")
+
+
+class AgentBuilder(A2AAgent[BuilderConfig, NoAuth]):
+    name = "agent-builder"
+    description = (
+        "Generate, test, and deploy new a2a-pack agents from natural language. "
+        "Writes the project into the user's workspace, validates it in a "
+        "sandbox, then ships it via the control plane."
+    )
+    version = "0.1.0"
+
+    config_model = BuilderConfig
+    auth_model = NoAuth
+    tools_used = ("deepagents", "langgraph", "a2a-pack", "litellm", "microsandbox")
+    llm_provisioning = LLMProvisioning.CALLER_PROVIDED
+    wants_cp_jwt = True
+    pricing = Pricing(
+        price_per_call_usd=0.10,
+        caller_pays_llm=True,
+        notes="LLM cost passes through; deployed agent is yours forever.",
+    )
+
+    @skill(
+        description=(
+            "Generate, test, and deploy a new a2a agent. Pass a kebab-case "
+            "name and a natural-language description of what the agent "
+            "should do. Returns the live URL when ready."
+        ),
+        tags=["builder", "scaffold", "deepagents", "meta"],
+    )
+    async def build(
+        self,
+        ctx: RunContext[NoAuth],
+        name: str,
+        prompt: str,
+        public: bool = True,
+        version: str = "0.1.0",
+    ) -> dict:
+        bucket = getattr(ctx.workspace, "bucket", None)
+        if not bucket:
+            return {"error": "no workspace grant; refusing to run"}
+        cp_jwt = ctx.cp_jwt
+        if not cp_jwt:
+            return {
+                "error": "no CP JWT forwarded — caller must run me through "
+                "the platform orchestrator so deploy can act on your behalf",
+            }
+        if not _is_valid_slug(name):
+            return {"error": f"invalid name {name!r} — use kebab-case, 2-63 chars"}
+
+        await ctx.emit_progress(f"building agent {name!r} in bucket {bucket}")
+        creds = ctx.llm
+        await ctx.emit_progress(
+            f"llm: {creds.model} via {creds.source} ({creds.base_url})"
+        )
+
+        graph = build_agent_builder(BuilderContext(
+            bucket=bucket, cp_jwt=cp_jwt,
+            llm_base_url=creds.base_url,
+            llm_api_key=creds.api_key,
+            llm_model=creds.model,
+        ))
+
+        user_msg = (
+            f"Build an a2a-pack agent named ``{name}``.\n\n"
+            f"The agent should: {prompt}\n\n"
+            f"After writing agent.py + a2a.yaml + requirements.txt, run "
+            f"test_agent_in_sandbox to verify it loads cleanly, then "
+            f"cp_deploy_tarball(name={name!r}, version={version!r}, "
+            f"public={public}). Report the URL the platform gives back."
+        )
+
+        final_state: dict[str, Any] = {}
+        deployed_url: str | None = None
+        last_reply: str = ""
+        try:
+            async for event in graph.astream_events(
+                {"messages": [{"role": "user", "content": user_msg}]},
+                version="v2",
+            ):
+                kind = event.get("event")
+                tname = event.get("name") or ""
+                data = event.get("data") or {}
+                if kind == "on_tool_start":
+                    raw = data.get("input") or {}
+                    if isinstance(raw, dict) and set(raw.keys()) == {"input"}:
+                        raw = raw["input"]
+                    await ctx.emit_progress(f"→ {tname}({_one_line(raw)})")
+                elif kind == "on_tool_end":
+                    summary = _summarize(tname, data.get("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"))
+                elif kind == "on_chain_end" and not event.get("parent_ids"):
+                    out = data.get("output")
+                    if isinstance(out, dict):
+                        final_state = out
+        except Exception as exc:  # noqa: BLE001
+            return {"error": f"build graph failed: {type(exc).__name__}: {exc}"}
+
+        for msg in final_state.get("messages") or []:
+            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)
+
+        if not deployed_url:
+            return {
+                "ok": False,
+                "name": name,
+                "reply": last_reply[:2000],
+                "warning": "build graph finished but no live URL was found; "
+                           "check progress events for errors.",
+            }
+
+        await ctx.emit_progress(f"deployed: {deployed_url}")
+        return {
+            "ok": True,
+            "name": name,
+            "version": version,
+            "url": deployed_url,
+            "workspace_dir": f"agents/{name}/",
+            "reply": last_reply[:2000],
+        }
+
+
+def _is_valid_slug(name: str) -> bool:
+    return bool(re.match(r"^[a-z][a-z0-9-]{1,62}$", name))
+
+
+def _one_line(args: Any, limit: int = 120) -> str:
+    if isinstance(args, dict):
+        bits = []
+        for k, v in args.items():
+            s = v if isinstance(v, str) else json.dumps(v)
+            if len(s) > 60:
+                s = s[:60] + "…"
+            bits.append(f"{k}={s}")
+        return ", ".join(bits)[:limit]
+    return (json.dumps(args) if not isinstance(args, str) else args)[:limit]
+
+
+def _summarize(name: str, output: Any) -> str:
+    if output is None:
+        return "ok"
+    if isinstance(output, str):
+        try:
+            parsed = json.loads(output)
+        except (json.JSONDecodeError, ValueError):
+            return output[:120] + ("…" if len(output) > 120 else "")
+        return _summarize(name, parsed)
+    if isinstance(output, dict):
+        if "error" in output:
+            return f"error: {str(output['error'])[:120]}"
+        if name == "write_agent_file":
+            return f"wrote {output.get('path')}"
+        if name == "list_agent_files":
+            return f"{len(output.get('files', []))} files"
+        if name == "test_agent_in_sandbox":
+            ec = output.get("exit_code")
+            return f"exit={ec}"
+        if name == "cp_deploy_tarball":
+            url = output.get("url") or ""
+            return f"shipped → {url}" if url else "shipped"
+        return "ok"
+    return str(output)[:120]
+
+
+def _find_url(value: Any) -> str | None:
+    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"):
+                return v
+        # JSON-stringified outputs from tools.
+        return _find_url(json.dumps(value))
+    if isinstance(value, str):
+        m = _URL_RE.search(value)
+        return m.group(0) if m else None
+    return None