From d6655639530b60726938a59ff38b2be302d42ef8 Mon Sep 17 00:00:00 2001 From: robert Date: Mon, 1 Jun 2026 22:47:40 -0300 Subject: [PATCH] Add conversational meta-agent compose tool --- agent.py | 2 +- agent_builder/builder.py | 40 ++++++++- agent_builder/tools.py | 121 +++++++++++++++++++++++++- requirements.txt | 2 +- tests/test_template_init.py | 169 +++++++++++++++++++++++++++++++++++- 5 files changed, 327 insertions(+), 7 deletions(-) diff --git a/agent.py b/agent.py index 63d5bf7..b975416 100644 --- a/agent.py +++ b/agent.py @@ -278,4 +278,4 @@ def _find_url(value: Any) -> str | None: m = _URL_RE.search(value) return m.group(0) if m else None return None -# rebuild against a2a-pack 0.1.41 for current runtime defaults +# rebuild against a2a-pack 0.1.47 for manifest-backed meta-agent defaults diff --git a/agent_builder/builder.py b/agent_builder/builder.py index 96a0e33..0eb8f67 100644 --- a/agent_builder/builder.py +++ b/agent_builder/builder.py @@ -136,9 +136,37 @@ runtime: resources: cpu: "2" memory: 2Gi - max_runtime_seconds: 900 + max_runtime_seconds: 900 ``` +When the user asks to compose existing agents, build a declarative meta-agent +manifest instead of hand-writing orchestration code. The source of truth is a +JSON object with ``composition`` plus optional ``goal`` and ``memory``: + +```json +{ + "composition": { + "sub_agents": [ + {"name": "writer-agent", "skills": ["draft"]}, + {"tag": "charting", "skills": ["render_chart"], "required": false} + ], + "max_nodes": 6, + "max_parallel": 2, + "max_replans": 1 + }, + "goal": { + "objective": "Ship a launch report", + "success_criteria": ["draft complete", "chart complete"] + }, + "memory": {"tiers": ["files", "kv"], "namespace": "launch-report"} +} +``` + +Deploy that manifest with ``cp_compose_meta_agent``. That endpoint validates +children against the registry, generates editable ``MetaAgent`` source, commits +it, and deploys it through the same GitOps path. Use this path for +meta-agents unless the user explicitly asks for custom source. + For render/media/data agents, commands that create user-visible files MUST run through ``await ctx.workspace_shell(...)`` or ``await ctx.workspace_python(...)``. The platform sandbox persists ``/workspace`` writes directly and captures changed @@ -202,6 +230,13 @@ Your tools: elsewhere; use force=True only after the user explicitly accepts replacing the current repo source. + - cp_compose_meta_agent(name, manifest_json, description="", version="0.1.0", + public=True, refresh_existing=False) + — create/deploy a manifest-backed + meta-agent that composes existing agents. + Use for "compose these agents toward this + goal" requests instead of hand-writing + orchestration source. - sync_agent_workspace_from_repo(name) — replace agents// in MinIO with the current managed repo source and @@ -234,6 +269,9 @@ Discipline: ``csv-sanitizer``). Class name is PascalCase from the slug. - For a new project, call init_agent_template first. Then read/edit the generated files instead of inventing boilerplate from memory. + - For a new meta-agent that only composes existing agents, call + cp_compose_meta_agent with a manifest. Do not scaffold a normal project + unless the user needs custom code beyond composition. - Ensure all three core files (agent.py, a2a.yaml, requirements.txt) exist before testing — partial scaffolds break ``a2a card``. - When the user asks for a usable app, dashboard, charting UI, workflow UI, diff --git a/agent_builder/tools.py b/agent_builder/tools.py index 9372cdf..0908942 100644 --- a/agent_builder/tools.py +++ b/agent_builder/tools.py @@ -3,6 +3,7 @@ - workspace file CRUD through the caller's scoped workspace grant - sandbox python (for `a2a card` / `a2a validate` round-trips) - cp_deploy_tarball (call /v1/agents/from-tarball on the user's behalf) + - cp_compose_meta_agent (call /v1/agents/compose with a manifest) """ from __future__ import annotations @@ -28,7 +29,7 @@ if TYPE_CHECKING: from .config import Settings -A2A_PACK_MIN_VERSION = "0.1.41" +A2A_PACK_MIN_VERSION = "0.1.47" @dataclass(frozen=True) @@ -579,6 +580,108 @@ def build_tools(ctx: ToolContext) -> list[Any]: out["last_seen_version"] = live_card.get("version") return json.dumps(out) + @tool + async def cp_compose_meta_agent( + name: str, + manifest_json: str, + description: str = "", + version: str = "0.1.0", + public: bool = True, + refresh_existing: bool = False, + ) -> str: + """Deploy a declarative meta-agent composition through the control plane. + + Use this when the user asks to compose existing agents toward a goal. + ``manifest_json`` must be a JSON object with ``composition`` and + optional ``goal`` / ``memory`` blocks. The control plane validates + referenced sub-agents and skills, generates editable MetaAgent source, + commits it to the user's managed source repo, and stamps the runtime + repo. This is the preferred path for meta-agents; do not hand-write + orchestration source when a manifest is enough. + """ + try: + prefix = _agent_prefix(name) + manifest = _parse_manifest_json(manifest_json) + except ValueError as exc: + return json.dumps({"error": str(exc)}) + if not ctx.cp_jwt: + return json.dumps({ + "error": "no CP JWT forwarded — agent declaration missing wants_cp_jwt=True", + }) + body = { + "name": name, + "description": description, + "version": version, + "public": public, + "manifest": manifest, + "refresh_existing": refresh_existing, + } + try: + async with httpx.AsyncClient(timeout=60.0) as c: + r = await c.post( + f"{settings.cp_url}/v1/agents/compose", + headers={"authorization": f"bearer {ctx.cp_jwt}"}, + json=body, + ) + 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]}) + payload = r.json() or {} + head_sha = payload.get("head_sha") + store.put( + prefix + "meta_agent_manifest.json", + (json.dumps(manifest, indent=2, sort_keys=True) + "\n").encode("utf-8"), + "application/json", + ) + if isinstance(head_sha, str) and head_sha: + _write_builder_state( + store, + name, + { + "agent": name, + "repo_head_sha": head_sha, + "deployment_id": payload.get("deployment_id"), + "version": payload.get("version"), + "updated_at": int(time.time()), + "source": "agent-builder-compose", + }, + ) + url = payload.get("expected_url") or payload.get("url") + live, live_card = await _wait_for_live_card( + url, + payload.get("version"), + timeout_s=settings.deploy_wait_timeout_s, + ) + out: dict[str, Any] = { + "ok": live, + "name": payload.get("name"), + "version": payload.get("version"), + "status": payload.get("status"), + "url": url, + "head_sha": head_sha, + "deployment_id": payload.get("deployment_id"), + "preview": payload.get("preview") or {}, + "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"] = ( + "compose deploy did not become live within timeout — build may " + "still be rolling. Inspect deployment_id or call cp_refresh_agent later." + ) + 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 @@ -759,7 +862,8 @@ def build_tools(ctx: ToolContext) -> list[Any]: return [ init_agent_template, list_agent_files, write_agent_file, read_agent_file, - write_agent_skill, test_agent_in_sandbox, cp_deploy_tarball, cp_refresh_agent, + write_agent_skill, test_agent_in_sandbox, cp_deploy_tarball, + cp_compose_meta_agent, cp_refresh_agent, sync_agent_workspace_from_repo, list_a2a_pack, read_a2a_pack, ] @@ -810,6 +914,19 @@ def _write_builder_state( store.put(_builder_state_key(name), body, "application/json") +def _parse_manifest_json(raw: str) -> dict[str, Any]: + try: + parsed = json.loads(raw or "{}") + except (TypeError, ValueError) as exc: + raise ValueError("manifest_json must be a JSON object") from exc + if not isinstance(parsed, dict): + raise ValueError("manifest_json must be a JSON object") + composition = parsed.get("composition") + if not isinstance(composition, (dict, list)): + raise ValueError("manifest_json requires a composition object or list") + return parsed + + def _render_skill_md(skill_name: str, description: str, instructions: str) -> str: desc = " ".join(description.strip().split()) body = instructions.strip() diff --git a/requirements.txt b/requirements.txt index c86ed2c..dafd889 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ -a2a-pack>=0.1.41 +a2a-pack>=0.1.47 httpx>=0.27 boto3>=1.34 deepagents>=0.5.0 diff --git a/tests/test_template_init.py b/tests/test_template_init.py index d0d5dac..be47a07 100644 --- a/tests/test_template_init.py +++ b/tests/test_template_init.py @@ -1,14 +1,17 @@ from __future__ import annotations +import asyncio import io import json import tarfile import unittest +from unittest.mock import patch from agent_builder.config import Settings from agent_builder.tools import ( _BUILDER_STATE_FILE, _BUILDER_INTERNAL_PREFIX, + _parse_manifest_json, _deployment_drift_error, _files_from_tarball, _parse_supporting_skill_files, @@ -235,6 +238,124 @@ class TemplateInitTests(unittest.TestCase): self.assertIn("agents/website-scraper/a2a.yaml", workspace.objects) self.assertIn("agents/website-scraper/requirements.txt", workspace.objects) + def test_parse_manifest_json_requires_object_with_composition(self) -> None: + with self.assertRaises(ValueError): + _parse_manifest_json("[]") + with self.assertRaises(ValueError): + _parse_manifest_json('{"goal":"ship"}') + + parsed = _parse_manifest_json('{"composition":[{"name":"writer"}]}') + self.assertEqual(parsed["composition"][0]["name"], "writer") + + def test_cp_compose_meta_agent_requires_manifest_object(self) -> None: + tools = build_tools( + ToolContext( + bucket="bucket", + settings=_settings(), + workspace=_FakeWorkspace(), + cp_jwt="jwt-user", + ) + ) + compose = _tool_by_name(tools, "cp_compose_meta_agent") + + result = json.loads( + asyncio.run( + compose.ainvoke( + { + "name": "report-meta", + "manifest_json": "[]", + } + ) + ) + ) + + self.assertIn("manifest_json must be a JSON object", result["error"]) + + def test_cp_compose_meta_agent_posts_manifest_and_records_state(self) -> None: + workspace = _FakeWorkspace() + tools = build_tools( + ToolContext( + bucket="bucket", + settings=_settings(deploy_wait_timeout_s=0), + workspace=workspace, + cp_jwt="jwt-user", + ) + ) + compose = _tool_by_name(tools, "cp_compose_meta_agent") + manifest = { + "composition": { + "sub_agents": [{"name": "writer", "skills": ["draft"]}], + "max_nodes": 3, + }, + "goal": { + "objective": "Ship a report.", + "success_criteria": ["draft complete"], + }, + "memory": {"tiers": ["files", "kv"], "namespace": "report"}, + } + + _FakeAsyncClient.posts = [] + with patch("agent_builder.tools.httpx.AsyncClient", _FakeAsyncClient): + result = json.loads( + asyncio.run( + compose.ainvoke( + { + "name": "report-meta", + "manifest_json": json.dumps(manifest), + "description": "Report coordinator", + "version": "1.2.3", + "public": True, + } + ) + ) + ) + + self.assertFalse(result["ok"]) + self.assertEqual(result["name"], "report-meta") + self.assertEqual(result["head_sha"], "source-sha") + self.assertEqual(len(_FakeAsyncClient.posts), 1) + post = _FakeAsyncClient.posts[0] + self.assertEqual(post["url"], "http://cp.test/v1/agents/compose") + self.assertEqual(post["headers"]["authorization"], "bearer jwt-user") + self.assertEqual(post["json"]["name"], "report-meta") + self.assertEqual(post["json"]["manifest"], manifest) + self.assertTrue(post["json"]["public"]) + + manifest_key = "agents/report-meta/meta_agent_manifest.json" + self.assertIn(manifest_key, workspace.objects) + saved_manifest = json.loads(workspace.objects[manifest_key].decode("utf-8")) + self.assertEqual(saved_manifest["goal"]["objective"], "Ship a report.") + state = json.loads( + workspace.objects[ + "agents/report-meta/" + _BUILDER_STATE_FILE + ].decode("utf-8") + ) + self.assertEqual(state["repo_head_sha"], "source-sha") + self.assertEqual(state["source"], "agent-builder-compose") + + def test_cp_compose_meta_agent_requires_cp_jwt(self) -> None: + tools = build_tools( + ToolContext( + bucket="bucket", + settings=_settings(), + workspace=_FakeWorkspace(), + ) + ) + compose = _tool_by_name(tools, "cp_compose_meta_agent") + + result = json.loads( + asyncio.run( + compose.ainvoke( + { + "name": "report-meta", + "manifest_json": '{"composition":[{"name":"writer"}]}', + } + ) + ) + ) + + self.assertIn("no CP JWT", result["error"]) + def _tarball(files: dict[str, str]) -> bytes: buf = io.BytesIO() @@ -315,12 +436,12 @@ def _tool_by_name(tools: list[object], name: str) -> object: raise AssertionError(f"tool not found: {name}") -def _settings() -> Settings: +def _settings(*, deploy_wait_timeout_s: int = 1) -> Settings: return Settings( sandbox_url="http://sandbox.test", sandbox_timeout_s=1, sandbox_token=None, - deploy_wait_timeout_s=1, + deploy_wait_timeout_s=deploy_wait_timeout_s, litellm_url="http://litellm.test", litellm_key="", litellm_model="gpt-test", @@ -330,3 +451,47 @@ def _settings() -> Settings: minio_secret_key="secret", image="python:3.11-slim", ) + + +class _FakeResponse: + def __init__(self, status_code: int, body: dict[str, object]) -> None: + self.status_code = status_code + self._body = body + self.text = json.dumps(body) + + def json(self) -> dict[str, object]: + return self._body + + +class _FakeAsyncClient: + posts: list[dict[str, object]] = [] + + def __init__(self, *args: object, **kwargs: object) -> None: + pass + + async def __aenter__(self) -> "_FakeAsyncClient": + return self + + async def __aexit__(self, *args: object) -> None: + return None + + async def post( + self, + url: str, + *, + headers: dict[str, str], + json: dict[str, object], + ) -> _FakeResponse: + self.posts.append({"url": url, "headers": headers, "json": json}) + return _FakeResponse( + 201, + { + "name": json["name"], + "version": json["version"], + "status": "building", + "expected_url": "https://report-meta.a2acloud.io", + "deployment_id": "dpl_123", + "head_sha": "source-sha", + "preview": {"skills": ["pursue"]}, + }, + )