tools: expose a2a_pack source to the inner LLM loop
All checks were successful
build / build (push) Successful in 3s
All checks were successful
build / build (push) Successful in 3s
list_a2a_pack(subdir) / read_a2a_pack(path) let the builder browse the installed SDK at runtime — the same package the scaffolded agents import. Path-traversal guarded; reads capped at 64 KiB. Why: the system prompt only carries an abridged example. When the LLM needed to recall what RunContext exposes (ctx.workspace API, ctx.ask vs ctx.collect, runtime.apt_packages…) it would guess. Now it can read the actual source. Prompt updated to point at the new tools. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -94,6 +94,23 @@ Your tools:
|
||||
- cp_deploy_tarball(name, version="0.1.0", public=True)
|
||||
— ship it to the platform; returns the
|
||||
public URL
|
||||
- cp_refresh_agent(name) — force the control plane to re-fetch
|
||||
the agent's live card after an
|
||||
out-of-band redeploy
|
||||
- list_a2a_pack(subdir="") — browse the installed a2a_pack SDK
|
||||
- read_a2a_pack(path) — read SDK source. Use these when
|
||||
you're unsure what's exposed. The
|
||||
code you scaffold runs on this
|
||||
exact SDK, so reading it is the
|
||||
authoritative reference — start
|
||||
with ``read_a2a_pack("agent.py")``
|
||||
for ``@skill`` semantics,
|
||||
``context.py`` for ``RunContext``
|
||||
(workspace, sandbox, emit_progress,
|
||||
ask, collect, request_scope),
|
||||
``runtime.py`` for ``AgentRuntime``
|
||||
fields (apt_packages, pricing,
|
||||
egress, etc.).
|
||||
|
||||
Discipline:
|
||||
|
||||
|
||||
@@ -352,9 +352,91 @@ def build_tools(ctx: ToolContext) -> list[Any]:
|
||||
"card": body.get("card") or {},
|
||||
})
|
||||
|
||||
@tool
|
||||
def list_a2a_pack(subdir: str = "") -> str:
|
||||
"""List ``.py`` files under the installed ``a2a_pack`` package.
|
||||
|
||||
Use this when you need to recall what the SDK actually exposes
|
||||
— what decorators exist, what ``RunContext`` methods are
|
||||
available, how ``WorkspaceClient`` is shaped, what the
|
||||
``AgentRuntime`` fields are, etc. The agent code you scaffold
|
||||
runs on this exact package, so reading it is the authoritative
|
||||
reference (the system-prompt examples are intentionally
|
||||
abridged).
|
||||
|
||||
``subdir`` is an optional path segment relative to the package
|
||||
root (``""``, ``"serve"``, ``"mcp"`` …). Returns JSON:
|
||||
``{root, files: [{path, size}]}``.
|
||||
"""
|
||||
try:
|
||||
import a2a_pack
|
||||
import os as _os
|
||||
base = _os.path.dirname(_os.path.abspath(a2a_pack.__file__))
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return json.dumps({"error": f"a2a_pack not importable: {exc}"})
|
||||
if subdir.startswith("/") or ".." in subdir.split("/"):
|
||||
return json.dumps({"error": "subdir must be relative; '..' disallowed"})
|
||||
root = _os.path.join(base, subdir) if subdir else base
|
||||
if not _os.path.commonpath([base, _os.path.abspath(root)]) == base:
|
||||
return json.dumps({"error": "subdir escapes package root"})
|
||||
if not _os.path.isdir(root):
|
||||
return json.dumps({"error": f"not a directory: {subdir}"})
|
||||
out: list[dict[str, Any]] = []
|
||||
for dirpath, _, filenames in _os.walk(root):
|
||||
for fname in filenames:
|
||||
if not fname.endswith(".py"):
|
||||
continue
|
||||
full = _os.path.join(dirpath, fname)
|
||||
rel = _os.path.relpath(full, base)
|
||||
try:
|
||||
size = _os.path.getsize(full)
|
||||
except OSError:
|
||||
continue
|
||||
out.append({"path": rel, "size": size})
|
||||
out.sort(key=lambda r: r["path"])
|
||||
return json.dumps({"root": base, "files": out})
|
||||
|
||||
@tool
|
||||
def read_a2a_pack(path: str) -> str:
|
||||
"""Read a file from the installed ``a2a_pack`` package.
|
||||
|
||||
Path is relative to the package root, e.g. ``agent.py``,
|
||||
``context.py``, ``runtime.py``, ``workspace.py``, ``card.py``,
|
||||
``mcp/server.py``. Returns JSON: ``{path, content}``. Capped
|
||||
at 64 KiB; reach for ``list_a2a_pack`` first if you need to
|
||||
chase a deeper file.
|
||||
"""
|
||||
try:
|
||||
import a2a_pack
|
||||
import os as _os
|
||||
base = _os.path.dirname(_os.path.abspath(a2a_pack.__file__))
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return json.dumps({"error": f"a2a_pack not importable: {exc}"})
|
||||
if path.startswith("/") or ".." in path.split("/"):
|
||||
return json.dumps({"error": "path must be relative; '..' disallowed"})
|
||||
full = _os.path.abspath(_os.path.join(base, path))
|
||||
if _os.path.commonpath([base, full]) != base:
|
||||
return json.dumps({"error": "path escapes package root"})
|
||||
if not _os.path.isfile(full):
|
||||
return json.dumps({"error": f"not a file: {path}"})
|
||||
try:
|
||||
with open(full, "r", encoding="utf-8") as fh:
|
||||
content = fh.read(64 * 1024 + 1)
|
||||
except OSError as exc:
|
||||
return json.dumps({"error": str(exc)})
|
||||
truncated = len(content) > 64 * 1024
|
||||
if truncated:
|
||||
content = content[: 64 * 1024]
|
||||
return json.dumps({
|
||||
"path": path,
|
||||
"content": content,
|
||||
"truncated": truncated,
|
||||
})
|
||||
|
||||
return [
|
||||
list_agent_files, write_agent_file, read_agent_file,
|
||||
test_agent_in_sandbox, cp_deploy_tarball, cp_refresh_agent,
|
||||
list_a2a_pack, read_a2a_pack,
|
||||
]
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user