Add conversational meta-agent compose tool
All checks were successful
build / build (push) Successful in 15s
All checks were successful
build / build (push) Successful in 15s
This commit is contained in:
2
agent.py
2
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
|
||||
|
||||
@@ -139,6 +139,34 @@ runtime:
|
||||
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/<name>/ 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,
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
a2a-pack>=0.1.41
|
||||
a2a-pack>=0.1.47
|
||||
httpx>=0.27
|
||||
boto3>=1.34
|
||||
deepagents>=0.5.0
|
||||
|
||||
@@ -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"]},
|
||||
},
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user