gitea_admin/agent-builder
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Teach builder to author DeepAgents skills
All checks were successful
build / build (push) Successful in 26s
Details

Browse Source
This commit is contained in:
robert
2026-05-19 09:39:29 -03:00
parent e1d9eab879
commit f6c612a071
10 changed files with 857 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
README.md
View File

@@ -7,10 +7,24 @@ a2a-pack agents from a natural-language description.
The single `build(name, prompt)` skill:
1. Builds an inner `deepagents` LangGraph configured with five tools:
1. Builds an inner `deepagents` LangGraph configured with project tools and
packaged builder skills:
- `deepagent-agent-design` — when to use DeepAgents skills, subagents,
workspace backends, and deterministic tools
- `a2apack-agent-authoring` — how to shape `agent.py`, public `@skill`
schemas, runtime metadata, pricing, resources, and auth
- `deepagents-implementation-patterns` — concrete `create_deep_agent`
wiring for caller LLMs, skills, tools, and subagents
- `workspace-artifact-safety` — workspace grants, artifacts, bounded
subprocesses, and scope expansion
- `agent-quality-review` — pre-sandbox and pre-deploy checks for generated
agents
- `list_agent_files(name)` / `write_agent_file(name, path, content)`
/ `read_agent_file(name, path)` — boto3 → MinIO, scoped to
`agents/<name>/` in the caller's workspace
- `write_agent_skill(...)` — creates valid DeepAgents
`skills/<skill-name>/SKILL.md` bundles for generated agents, so rich
behavior lives in progressive-disclosure skills instead of fake tools
- `test_agent_in_sandbox(name)` — tarballs the workspace project,
spins a microVM with the public `a2a-pack` wheel installed, runs
`a2a card` to verify the scaffold is valid

68
agent_builder/builder.py
View File

@@ -2,15 +2,23 @@
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
from deepagents.backends.utils import create_file_data
from langchain_openai import ChatOpenAI
from langgraph.store.memory import InMemoryStore
from .config import Settings, load_settings
from .tools import ToolContext, build_tools
BUILDER_SKILL_SOURCE = "/.agent-builder/skills/"
_BUILDER_SKILL_NAMESPACE = ("agent-builder", "skills")
@dataclass(frozen=True)
class BuilderContext:
bucket: str
@@ -38,6 +46,13 @@ small tool-calling DeepAgent with ``create_deep_agent`` plus
``wrap_model_call`` middleware. Do not recreate this from memory: call
``init_agent_template`` first and modify the generated files.
You have packaged DeepAgents skills loaded from ``/.agent-builder/skills/``.
Use ``deepagent-agent-design`` before designing generated agent internals,
``a2apack-agent-authoring`` when shaping the public A2A class and Card,
``deepagents-implementation-patterns`` when wiring tools/subagents/skills,
``workspace-artifact-safety`` when files or artifacts are involved, and
``agent-quality-review`` before the final sandbox test/deploy pass.
The generated ``a2a.yaml`` looks like:
```yaml
@@ -107,6 +122,13 @@ Your tools:
- write_agent_file(name, path, content)
— save agent.py / a2a.yaml / requirements.txt
- read_agent_file(name, path) — re-read a file (for iteration)
- write_agent_skill(name, skill_name, description, instructions,
supporting_files_json="{}")
— create ``skills/<skill_name>/SKILL.md``
plus optional bundled resources for a
generated DeepAgent. Use this for rich
behavior before wiring ``skills=[...]``
in agent.py.
- test_agent_in_sandbox(name) — pip install + ``a2a card`` round-trip;
check exit_code == 0 and the card JSON
looks right
@@ -150,6 +172,14 @@ Discipline:
generated files instead of inventing boilerplate from memory.
- Ensure all three core files (agent.py, a2a.yaml, requirements.txt)
exist before testing — partial scaffolds break ``a2a card``.
- For non-trivial agents, create one or more DeepAgents skills under
``skills/<skill-name>/`` and wire ``create_deep_agent(..., skills=[...])``.
Do not replace LLM reasoning with fake deterministic tools that return
canned answers. Tools should do exact work; skills and subagents should
carry reusable workflow knowledge and judgement-heavy behavior.
- If generated agent.py calls ``ctx.workspace_backend()``, declare
``workspace_access = WorkspaceAccess.dynamic(...)`` on the A2A agent class
so local/dev/platform invocations actually receive a workspace grant.
- Always run test_agent_in_sandbox before deploying. If exit_code != 0,
read stderr, edit the offending file, retest. Do NOT deploy a broken
scaffold.
@@ -188,6 +218,7 @@ def build_agent_builder(ctx: BuilderContext) -> Any:
backend = _build_project_backend(ctx, settings=settings)
if backend is not None:
kwargs["backend"] = backend
kwargs["skills"] = [BUILDER_SKILL_SOURCE]
return create_deep_agent(**kwargs)
@@ -199,16 +230,16 @@ def _build_project_backend(ctx: BuilderContext, *, settings: Settings) -> Any |
LangGraph state. Scope it to the project prefix, not the whole bucket.
"""
if not ctx.project_prefix:
return None
return _with_builder_skills(StateBackend())
prefix = ctx.project_prefix.strip("/")
if not prefix:
return None
return _with_builder_skills(StateBackend())
try:
from a2a_pack import Grant, WorkspaceAccess, WorkspaceMode
from a2a_pack.deepagents import WorkspaceBackend
from a2a_pack.workspace import MinIOWorkspaceClient
except Exception: # noqa: BLE001
return None
return _with_builder_skills(StateBackend())
workspace = MinIOWorkspaceClient(
bucket=ctx.bucket,
@@ -234,4 +265,33 @@ def _build_project_backend(ctx: BuilderContext, *, settings: Settings) -> Any |
outputs_prefix=None,
)
)
return WorkspaceBackend(workspace)
return _with_builder_skills(WorkspaceBackend(workspace))
def _with_builder_skills(default_backend: Any) -> Any:
skill_store = InMemoryStore()
for path, content in _builder_skill_files().items():
skill_store.put(
_BUILDER_SKILL_NAMESPACE,
"/" + path,
create_file_data(content),
)
skill_backend = StoreBackend(
store=skill_store,
namespace=lambda _runtime: _BUILDER_SKILL_NAMESPACE,
)
return CompositeBackend(
default=default_backend,
routes={BUILDER_SKILL_SOURCE: skill_backend},
)
def _builder_skill_files() -> dict[str, str]:
root = Path(__file__).with_name("skills")
files: dict[str, str] = {}
if not root.exists():
return files
for path in root.rglob("*"):
if path.is_file():
files[path.relative_to(root).as_posix()] = path.read_text(encoding="utf-8")
return files

149
agent_builder/skills/a2apack-agent-authoring/SKILL.md Normal file
View File

@@ -0,0 +1,149 @@
---
name: a2apack-agent-authoring
description: Author production a2a-pack agents with clear public @skill schemas, runtime declarations, caller-provided LLMs, workspace grants, pricing, resources, and secure platform capabilities. Use when writing or reviewing agent.py for an A2A agent.
---
# A2A Pack Agent Authoring
An A2A agent should expose a small, stable public API and put rich reasoning
inside implementation helpers, DeepAgents skills, or subagents. The public
`@skill` methods are the contract shown on the Agent Card.
## Public Contract
Use `A2AAgent` class attributes for runtime and marketplace behavior:
- `name`, `description`, and `version` identify the agent.
- `config_model` and `auth_model` define agent configuration and caller auth.
- `llm_provisioning` tells the platform where LLM credentials come from.
- `pricing` tells callers what they pay and whether they supply LLM creds.
- `resources`, `egress`, `tools_used`, and `workspace_access` describe runtime
requirements.
- `wants_cp_jwt=True` is only for trusted platform agents. It forwards the
caller's control-plane token.
Keep each `@skill` method `async`, put `RunContext[...]` immediately after
`self`, and annotate every public argument. Do not use `*args` or `**kwargs`;
they are rejected and would not publish a useful schema.
## Code Example
```python
from __future__ import annotations
from typing import Any
from pydantic import BaseModel
from a2a_pack import (
A2AAgent,
EgressPolicy,
LLMProvisioning,
NoAuth,
Pricing,
Resources,
RunContext,
WorkspaceAccess,
WorkspaceMode,
skill,
)
class ResearchConfig(BaseModel):
default_depth: int = 3
class ResearchAgent(A2AAgent[ResearchConfig, NoAuth]):
name = "research-agent"
description = "Researches a topic and writes a concise cited report."
version = "0.1.0"
config_model = ResearchConfig
auth_model = NoAuth
llm_provisioning = LLMProvisioning.CALLER_PROVIDED
pricing = Pricing(
price_per_call_usd=0.05,
caller_pays_llm=True,
notes="Caller supplies LLM credentials through the platform.",
)
resources = Resources(cpu="1", memory="1Gi", max_runtime_seconds=900)
egress = EgressPolicy(allow_hosts=("api.openai.com",))
tools_used = ("deepagents", "langchain")
workspace_access = WorkspaceAccess.dynamic(
max_files=128,
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
require_reason=False,
)
@skill(
description="Research a topic, save report artifacts, and return a summary",
timeout_seconds=900,
cost_class="llm-heavy",
)
async def research(
self,
ctx: RunContext[NoAuth],
topic: str,
depth: int | None = None,
save_path: str = "outputs/research-report.md",
) -> dict[str, Any]:
creds = ctx.llm
await ctx.emit_progress(f"researching with {creds.model}")
graph = self._build_graph(ctx)
state = await graph.ainvoke(
{
"messages": [
{
"role": "user",
"content": (
f"Research {topic!r} at depth {depth or self.config.default_depth}. "
f"Write the final markdown report to /workspace/{save_path}."
),
}
]
}
)
report_text = _last_message_text(state)
ref = await ctx.write_artifact(
"research-summary.md",
report_text.encode("utf-8"),
"text/markdown",
)
await ctx.emit_artifact(ref)
return {"summary": report_text, "artifact": ref.uri, "path": save_path}
```
## Schema Rules
Public arguments become the live A2A input schema. Prefer plain JSON-shaped
types: `str`, `int`, `float`, `bool`, `list[str]`, `dict[str, Any]`, or
Pydantic models. Provide defaults for optional knobs. Avoid exposing internal
implementation details such as model names, temporary paths, prompt fragments,
or unbounded command strings.
## Runtime Rules
If a skill calls `ctx.workspace_backend()` or `ctx.workspace`, declare
`workspace_access`. If it writes user-visible files, also emit artifacts with
`ctx.write_artifact` and `ctx.emit_artifact`.
If an agent needs system binaries, mirror the need in `a2a.yaml`:
```yaml
runtime:
apt_packages: [ffmpeg, poppler-utils]
resources:
cpu: "2"
memory: 2Gi
max_runtime_seconds: 900
```
If it calls external hosts, declare `egress` on the class. If it needs secrets,
use `required_secrets` and `ctx.secret("NAME")`; do not hard-code credentials.
## A2A Skill vs DeepAgents Skill
Use A2A `@skill` for caller-visible operations and billing boundaries. Use
DeepAgents skills for internal procedural knowledge the LLM should apply. A
generated project can have one public `@skill` that invokes a DeepAgent loaded
with several internal `skills/<name>/SKILL.md` bundles.

80
agent_builder/skills/agent-quality-review/SKILL.md Normal file
View File

@@ -0,0 +1,80 @@
---
name: agent-quality-review
description: Review a generated a2a-pack agent before sandbox testing or deploy. Use to catch broken cards, missing schemas, fake tools, unwired DeepAgents skills, missing runtime mirrors, unsafe IO, and stale live Agent Card schemas.
---
# Agent Quality Review
Before `test_agent_in_sandbox` and before deploy, review the generated project
as source code, not just as text.
## Required Files
Every generated project needs:
- `agent.py`
- `a2a.yaml`
- `requirements.txt`
- Optional `skills/<skill-name>/SKILL.md` bundles for nontrivial DeepAgents
behavior
Run `list_agent_files` and read the files that matter. If the project was
deployed before, use `sync_agent_workspace_from_repo` before editing when the
builder needs to continue from the current managed repo source.
## Agent.py Review
Check these items:
- Public `@skill` methods are async and have `RunContext[...]` plus typed JSON
arguments.
- The public schema is small and stable. No unbounded command strings, hidden
prompt fragments, or provider credentials as public inputs.
- The implementation reads `ctx.llm` when `llm_provisioning` is
`CALLER_PROVIDED`.
- Any use of DeepAgents file tools passes `backend=ctx.workspace_backend()`.
- Any project skills are seeded into the backend and passed with
`skills=skill_sources or None`.
- Custom subagents that need project skills include their own `skills` field.
- Deterministic tools do exact work; no fake canned-response tools.
- File-producing skills call `ctx.write_artifact` and `ctx.emit_artifact`.
- External calls, secrets, workspace access, pricing, and resources are
declared on the class.
## A2A YAML Review
Mirror deployment-only details in `a2a.yaml` because the platform reads it
before importing user code:
```yaml
name: research-agent
version: 0.1.0
entrypoint: agent:ResearchAgent
expose:
public: true
runtime:
apt_packages: [ffmpeg]
resources:
cpu: "2"
memory: 2Gi
max_runtime_seconds: 900
```
Use `runtime.apt_packages` only for Debian system binaries. Python packages
belong in `requirements.txt`.
## Sandbox And Live Card
Always run `test_agent_in_sandbox(name)` before deploy. Treat nonzero exit as
source failure, then read stderr, edit, and rerun.
After `cp_deploy_tarball`, compare `live_skills[].input_schema` with the
actual public `@skill` signatures. If a live schema is missing an argument or
shows an old shape, refresh or redeploy until the live Agent Card matches the
source.
## Minimal Good Result
For most generated agents, a good result is one public A2A `@skill`, one
workspace-backed DeepAgent, one or two internal DeepAgents skills, and a few
small deterministic tools. Resist broad endpoint sets unless the user asked
for a multi-operation agent.

117
agent_builder/skills/deepagent-agent-design/SKILL.md Normal file
View File

@@ -0,0 +1,117 @@
---
name: deepagent-agent-design
description: Design generated A2A agents around DeepAgents skills, subagents, durable workspace backends, and caller-provided LLM reasoning. Use whenever building or modifying an agent, deciding whether to add tools, creating skill bundles, wiring create_deep_agent, or avoiding shallow fake tools.
---
# DeepAgent Agent Design
Build generated agents as small A2A surfaces around a capable inner DeepAgent.
The A2A `@skill` is the user-facing API entrypoint. The inner DeepAgent does
planning, file work, skill selection, and subagent delegation.
## Default Architecture
Use this shape for non-trivial generated agents:
1. Keep one or two public A2A `@skill` methods with clear typed parameters.
2. Inside each method, read `ctx.llm`, construct `ChatOpenAI`, and build a
DeepAgent with `create_deep_agent`.
3. Pass `backend=ctx.workspace_backend()` so DeepAgents file tools write to the
caller's durable workspace instead of LangGraph state.
4. Pass `skills=[RUNTIME_SKILLS_ROOT]` when the generated project includes
`skills/<skill-name>/SKILL.md`.
5. For custom subagents, include a `skills` field on each subagent definition.
The general-purpose subagent inherits main skills, but custom subagents do
not.
DeepAgents skills are progressive-disclosure folders:
```text
skills/
skill-name/
SKILL.md
references/...
scripts/...
assets/...
```
`SKILL.md` must start with YAML frontmatter containing `name` and
`description`. The description is the trigger; include exactly when the skill
should be used. Keep detailed reference material in separate files that are
linked from `SKILL.md`.
## Runtime Skill Seeding
Project `skills/` files are source files in the agent image. DeepAgents loads
skills from its backend, so seed those files into the invocation workspace
before `create_deep_agent`:
```python
from pathlib import Path
RUNTIME_SKILLS_ROOT = "/outputs/{{ agent_name }}/.deepagents/skills/"
def _seed_runtime_skills(backend: Any) -> None:
root = Path(__file__).parent / "skills"
if not root.exists():
return
uploads: list[tuple[str, bytes]] = []
for path in root.rglob("*"):
if path.is_file():
rel = path.relative_to(root).as_posix()
uploads.append((RUNTIME_SKILLS_ROOT + rel, path.read_bytes()))
if uploads:
backend.upload_files(uploads)
```
Then:
```python
backend = ctx.workspace_backend()
_seed_runtime_skills(backend)
return create_deep_agent(
model=model,
backend=backend,
skills=[RUNTIME_SKILLS_ROOT],
tools=[small_deterministic_tool],
subagents=[...],
system_prompt=SYSTEM_PROMPT,
)
```
Because this uses `/outputs/...`, the write fits the normal A2A workspace
grant. If the agent uses `ctx.workspace_backend()`, also declare
`workspace_access = WorkspaceAccess.dynamic(...)` on the A2A agent class.
## Tool Design Rules
Do not make a long list of fake tools that just return canned text or call the
LLM again. Prefer:
- DeepAgents skills for procedural/domain knowledge the LLM should apply.
- Subagents for independent research, analysis, review, or transformation
lanes that benefit from another LLM pass.
- Small deterministic tools only for exact operations: parsing, validation,
API calls, math, file conversion, sandbox commands, database queries.
If a tool's body is mostly prompt text, it should usually be a DeepAgents
skill instead. If a tool needs judgement, route that judgement through the
inner DeepAgent or a subagent, not a hard-coded placeholder.
## Skill Authoring Workflow
When the generated agent needs reusable workflow knowledge, first call
`write_agent_skill` to create `skills/<skill-name>/SKILL.md` and any support
files. Then wire `agent.py` to seed and pass those skills.
Use concise skill descriptions with concrete trigger phrases. Example:
```yaml
---
name: market-research
description: Plan and run multi-source market research, delegate subtopics to subagents, save findings files, and synthesize cited reports. Use for competitive analysis, market maps, customer research, or current market questions.
---
```
Keep A2A public schemas stable and simple. Put rich autonomous behavior behind
the inner DeepAgent, not in a pile of public endpoints.

129
agent_builder/skills/deepagents-implementation-patterns/SKILL.md Normal file
View File

@@ -0,0 +1,129 @@
---
name: deepagents-implementation-patterns
description: Implement DeepAgents inside a2a-pack agents with real deterministic tools, project skills, custom subagents, workspace-backed files, and caller-provided ChatOpenAI models. Use when editing create_deep_agent calls or deciding what belongs in tools, skills, or subagents.
---
# DeepAgents Implementation Patterns
DeepAgents supplies the inner agent loop. In this platform, the reliable shape
is: caller-provided `ChatOpenAI`, A2A workspace backend, project DeepAgents
skills, and a small number of exact tools.
## Implementation Shape
```python
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from deepagents import create_deep_agent
RUNTIME_SKILLS_ROOT = "/outputs/research-agent/.deepagents/skills/"
SYSTEM_PROMPT = """\
You are a careful research agent. Use project skills for workflow rules and
write durable files under /workspace/outputs/.
"""
def build_graph(ctx: Any) -> Any:
creds = ctx.llm
model = ChatOpenAI(
model=creds.model,
base_url=creds.base_url,
api_key=creds.api_key,
temperature=0.0,
)
@tool
def validate_citations(items_json: str) -> str:
"""Validate that each citation has title, url, and claim fields."""
items = json.loads(items_json)
missing = [
i for i, item in enumerate(items)
if not {"title", "url", "claim"} <= set(item)
]
return json.dumps({"ok": not missing, "missing_indexes": missing})
backend = ctx.workspace_backend()
skill_sources = seed_runtime_skills(backend)
return create_deep_agent(
model=model,
backend=backend,
skills=skill_sources or None,
tools=[validate_citations],
subagents=[
{
"name": "source-reviewer",
"description": "Reviews source quality and citation coverage.",
"system_prompt": (
"Check whether sources support the claims. Return gaps "
"and concrete fixes."
),
"skills": skill_sources,
}
],
system_prompt=SYSTEM_PROMPT,
)
def seed_runtime_skills(backend: Any) -> list[str]:
root = Path(__file__).parent / "skills"
if not root.exists():
return []
uploads: list[tuple[str, bytes]] = []
for path in root.rglob("*"):
if path.is_file():
rel = path.relative_to(root).as_posix()
uploads.append((RUNTIME_SKILLS_ROOT + rel, path.read_bytes()))
if uploads:
backend.upload_files(uploads)
return [RUNTIME_SKILLS_ROOT]
return []
```
## Choose The Right Primitive
Use a DeepAgents skill when the agent needs reusable judgement, process, domain
rules, examples, or a checklist. Put it under `skills/<skill-name>/SKILL.md`.
Use a subagent when a separate LLM pass can independently research, review,
transform, or critique work. Custom subagents should receive their own `skills`
field; do not assume they inherit every main-agent skill.
Use a deterministic tool only for exact work: JSON validation, schema checks,
filesystem transforms, API calls, calculations, sandbox commands, conversion
commands, or database queries.
Do not write tools whose body is prompt text, canned answers, or another hidden
LLM call. That logic belongs in a DeepAgents skill or a subagent.
## Skill Files
Each skill directory must contain `SKILL.md` with frontmatter:
```markdown
---
name: market-research
description: Run market research with source review, competitor mapping, and cited synthesis. Use for market questions, competitive analysis, or customer research.
---
# Market Research
Follow this workflow...
```
The `description` is the trigger text visible to the LLM before full skill
loading. Keep it concrete. Put long examples or data dictionaries into
`references/`, scripts into `scripts/`, and reusable assets into `assets/`.
## Backend Rules
DeepAgents' default state backend is not enough for hosted agents because file
outputs can disappear into graph state. Always pass `backend=ctx.workspace_backend()`
for generated agents that read or write files.
Use `skills=skill_sources or None`, not an empty list, so agents without a
`skills/` directory still build cleanly.

113
agent_builder/skills/workspace-artifact-safety/SKILL.md Normal file
View File

@@ -0,0 +1,113 @@
---
name: workspace-artifact-safety
description: Write A2A agents that use ctx.workspace_backend, workspace grants, artifacts, sandbox commands, and file paths safely. Use for agents that read files, create outputs, run subprocesses, convert media, or return downloadable artifacts.
---
# Workspace And Artifact Safety
Generated agents run with explicit workspace grants. File work must fit those
grants and produce artifacts callers can find.
## Workspace Access
If `agent.py` uses `ctx.workspace`, `ctx.workspace_backend()`, or a DeepAgents
graph with file tools, declare workspace access on the class:
```python
from a2a_pack import WorkspaceAccess, WorkspaceMode
workspace_access = WorkspaceAccess.dynamic(
max_files=128,
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
require_reason=False,
)
```
Pass the backend into DeepAgents:
```python
backend = ctx.workspace_backend()
graph = create_deep_agent(model=model, backend=backend, skills=skill_sources or None)
```
Write durable model-created files under `/workspace/outputs/...` in prompts.
When writing through `ctx.workspace`, use workspace-relative paths such as
`outputs/report.md`.
## Artifact Pattern
For user-visible outputs, return a small JSON object and emit an artifact:
```python
async def save_report(ctx: RunContext[NoAuth], report: str) -> dict[str, str]:
ref = await ctx.write_artifact(
"report.md",
report.encode("utf-8"),
"text/markdown",
)
await ctx.emit_artifact(ref)
return {"artifact_uri": ref.uri, "name": ref.name}
```
For binary files, use the correct media type:
```python
ref = await ctx.write_artifact("frames.zip", zip_bytes, "application/zip")
await ctx.emit_artifact(ref)
```
## Bounded Subprocesses
Media, rendering, archive, and data conversion agents must not run unbounded
commands. Use explicit timeouts and return structured errors:
```python
import asyncio
async def run_checked(cmd: list[str], timeout_s: int = 120) -> dict[str, str | int]:
try:
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout_s)
except asyncio.TimeoutError:
return {"ok": 0, "error": "command timed out", "timeout_s": timeout_s}
return {
"ok": 1 if proc.returncode == 0 else 0,
"returncode": proc.returncode,
"stdout": stdout.decode("utf-8", errors="replace")[-4000:],
"stderr": stderr.decode("utf-8", errors="replace")[-4000:],
}
```
If a nonzero command still produced a usable output, preserve the output and
include the command failure as a warning instead of discarding the file.
## Scope Expansion
Only use `ctx.request_scope(...)` when the public `@skill` declares
`allow_scope_expansion=True`. Explain the reason in plain language and keep
the requested read/write patterns narrow.
```python
@skill(
description="Analyze a protected folder after the caller approves access",
allow_scope_expansion=True,
)
async def analyze_folder(ctx: RunContext[NoAuth], folder: str) -> dict[str, str]:
await ctx.request_scope(
reason=f"Need to read {folder} to analyze the requested files",
read=(f"{folder.rstrip('/')}/**",),
mode="read_only",
)
return {"status": "approved"}
```
## Safety Checklist
Do not hard-code absolute local paths. Do not expose arbitrary shell command
strings as public inputs. Do not write secrets to artifacts. Do not use
`wants_cp_jwt=True` for normal agents. Do not assume a file exists; check and
return a structured error that the caller can act on.

130
agent_builder/tools.py
View File

@@ -92,7 +92,9 @@ def _ensure_bucket(s3: Any, bucket: str) -> None:
_AGENT_NAME_RE = re.compile(r"^[a-z][a-z0-9-]{1,62}$")
_SKILL_NAME_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,62}[a-z0-9])?$")
_BUILDER_STATE_FILE = ".a2a-builder-state.json"
_BUILDER_INTERNAL_PREFIX = ".agent-builder/"
def _validate_name(name: str) -> None:
@@ -102,6 +104,14 @@ def _validate_name(name: str) -> None:
)
def _validate_skill_name(name: str) -> None:
if not _SKILL_NAME_RE.match(name) or "--" in name:
raise ValueError(
"invalid skill name "
f"{name!r}: use lowercase alphanumeric kebab-case, max 64 chars"
)
def _agent_prefix(name: str) -> str:
_validate_name(name)
return f"agents/{name}/"
@@ -162,7 +172,7 @@ def build_tools(ctx: ToolContext) -> list[Any]:
for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
for obj in page.get("Contents") or ():
rel = obj["Key"][len(prefix):]
if rel == _BUILDER_STATE_FILE:
if rel == _BUILDER_STATE_FILE or rel.startswith(_BUILDER_INTERNAL_PREFIX):
continue
out.append({
"path": rel,
@@ -182,8 +192,8 @@ def build_tools(ctx: ToolContext) -> list[Any]:
except ValueError as exc:
return json.dumps({"error": str(exc)})
rel = path.lstrip("/")
if rel == _BUILDER_STATE_FILE:
return json.dumps({"error": f"{_BUILDER_STATE_FILE} is managed by deploys"})
if rel == _BUILDER_STATE_FILE or rel.startswith(_BUILDER_INTERNAL_PREFIX):
return json.dumps({"error": f"{rel} is managed by agent-builder"})
key = prefix + rel
s3.put_object(
Bucket=bucket, Key=key, Body=content.encode("utf-8"),
@@ -199,8 +209,8 @@ def build_tools(ctx: ToolContext) -> list[Any]:
except ValueError as exc:
return json.dumps({"error": str(exc)})
rel = path.lstrip("/")
if rel == _BUILDER_STATE_FILE:
return json.dumps({"error": f"{_BUILDER_STATE_FILE} is managed by deploys"})
if rel == _BUILDER_STATE_FILE or rel.startswith(_BUILDER_INTERNAL_PREFIX):
return json.dumps({"error": f"{rel} is managed by agent-builder"})
key = prefix + rel
try:
obj = s3.get_object(Bucket=bucket, Key=key)
@@ -209,6 +219,59 @@ def build_tools(ctx: ToolContext) -> list[Any]:
body = obj["Body"].read().decode("utf-8", errors="replace")
return json.dumps({"path": key, "content": body[:200_000]})
@tool
def write_agent_skill(
name: str,
skill_name: str,
description: str,
instructions: str,
supporting_files_json: str = "{}",
) -> str:
"""Create a DeepAgents skill bundle in ``agents/<name>/skills/``.
Use this for generated agents that should rely on DeepAgents'
progressive-disclosure skills instead of fake one-off tools.
``supporting_files_json`` is a JSON object mapping relative paths
inside the skill directory to text content, for example
``{"references/schema.md": "..."}``.
After writing skills, update ``agent.py`` so it seeds packaged
``skills/`` files into ``ctx.workspace_backend()`` and passes
``skills=[...]`` to ``create_deep_agent``.
"""
try:
prefix = _agent_prefix(name)
_validate_skill_name(skill_name)
files = _parse_supporting_skill_files(supporting_files_json)
skill_md = _render_skill_md(skill_name, description, instructions)
except ValueError as exc:
return json.dumps({"error": str(exc)})
base = prefix + f"skills/{skill_name}/"
written: list[dict[str, Any]] = []
payloads = {"SKILL.md": skill_md, **files}
for rel, content in payloads.items():
key = base + rel
encoded = content.encode("utf-8")
s3.put_object(
Bucket=bucket,
Key=key,
Body=encoded,
ContentType="text/plain; charset=utf-8",
)
written.append({"path": key, "size": len(encoded)})
return json.dumps({
"ok": True,
"agent": name,
"skill": skill_name,
"source_path": f"skills/{skill_name}/",
"files": written,
"wire_agent_py": (
"Seed project skills into ctx.workspace_backend() and pass "
"skills=[RUNTIME_SKILLS_ROOT] to create_deep_agent."
),
})
@tool
async def test_agent_in_sandbox(name: str) -> str:
"""Spin a microVM, ``pip install a2a-pack`` + the agent's own
@@ -583,7 +646,7 @@ def build_tools(ctx: ToolContext) -> list[Any]:
return [
init_agent_template, list_agent_files, write_agent_file, read_agent_file,
test_agent_in_sandbox, cp_deploy_tarball, cp_refresh_agent,
write_agent_skill, test_agent_in_sandbox, cp_deploy_tarball, cp_refresh_agent,
sync_agent_workspace_from_repo,
list_a2a_pack, read_a2a_pack,
]
@@ -605,7 +668,7 @@ def _tarball_workspace_dir(s3: Any, bucket: str, prefix: str) -> bytes:
rel = key[len(prefix):]
if not rel:
continue
if rel == _BUILDER_STATE_FILE:
if rel == _BUILDER_STATE_FILE or rel.startswith(_BUILDER_INTERNAL_PREFIX):
continue
body = s3.get_object(Bucket=bucket, Key=key)["Body"].read()
info = tarfile.TarInfo(name=rel)
@@ -646,6 +709,59 @@ def _write_builder_state(
)
def _render_skill_md(skill_name: str, description: str, instructions: str) -> str:
desc = " ".join(description.strip().split())
body = instructions.strip()
if not desc:
raise ValueError("skill description is required")
if len(desc) > 1024:
raise ValueError("skill description must be 1024 characters or less")
if not body:
raise ValueError("skill instructions are required")
if "\n---" in body or body.startswith("---"):
raise ValueError("skill instructions must not contain frontmatter delimiters")
return (
"---\n"
f"name: {skill_name}\n"
"description: "
+ json.dumps(desc)
+ "\n"
"---\n"
f"# {skill_name}\n\n"
+ body
+ "\n"
)
def _parse_supporting_skill_files(raw: str) -> dict[str, str]:
try:
parsed = json.loads(raw or "{}")
except (TypeError, ValueError) as exc:
raise ValueError("supporting_files_json must be a JSON object") from exc
if not isinstance(parsed, dict):
raise ValueError("supporting_files_json must be a JSON object")
out: dict[str, str] = {}
for path, content in parsed.items():
if not isinstance(path, str) or not isinstance(content, str):
raise ValueError("supporting skill file paths and contents must be strings")
rel = _safe_supporting_skill_path(path)
if rel == "SKILL.md":
raise ValueError("supporting_files_json cannot override SKILL.md")
out[rel] = content
return out
def _safe_supporting_skill_path(path: str) -> str:
rel = path.replace("\\", "/").lstrip("/")
while rel.startswith("./"):
rel = rel[2:]
if not rel or ".." in rel.split("/") or rel.startswith("."):
raise ValueError(f"unsafe supporting skill file path: {path!r}")
if rel.endswith("/") or "//" in rel:
raise ValueError(f"unsafe supporting skill file path: {path!r}")
return rel
def _delete_workspace_objects(s3: Any, bucket: str, prefix: str) -> None:
paginator = s3.get_paginator("list_objects_v2")
batch: list[dict[str, str]] = []

38
tests/test_builder_prompt.py
View File

@@ -2,7 +2,13 @@ from __future__ import annotations
import unittest
from agent_builder.builder import SYSTEM_PROMPT
from agent_builder.builder import (
BUILDER_SKILL_SOURCE,
SYSTEM_PROMPT,
_builder_skill_files,
_with_builder_skills,
)
from deepagents.backends import StateBackend
class BuilderPromptTests(unittest.TestCase):
@@ -17,3 +23,33 @@ class BuilderPromptTests(unittest.TestCase):
self.assertIn("class GeneratedDeck(Slide)", SYSTEM_PROMPT)
self.assertIn("manim-slides render --CE -ql", SYSTEM_PROMPT)
self.assertIn("render: bool = False", SYSTEM_PROMPT)
def test_prompt_and_backend_load_packaged_builder_skills(self) -> None:
expected = {
"deepagent-agent-design",
"a2apack-agent-authoring",
"deepagents-implementation-patterns",
"workspace-artifact-safety",
"agent-quality-review",
}
for skill_name in expected:
self.assertIn(skill_name, SYSTEM_PROMPT)
self.assertIn("deepagent-agent-design", SYSTEM_PROMPT)
self.assertIn("write_agent_skill", SYSTEM_PROMPT)
self.assertIn("workspace_access = WorkspaceAccess.dynamic", SYSTEM_PROMPT)
files = _builder_skill_files()
self.assertTrue(
{path.split("/", 1)[0] for path in files}.issuperset(expected)
)
self.assertIn("skills=[RUNTIME_SKILLS_ROOT]", files["deepagent-agent-design/SKILL.md"])
self.assertIn("from a2a_pack import", files["a2apack-agent-authoring/SKILL.md"])
self.assertIn("create_deep_agent", files["deepagents-implementation-patterns/SKILL.md"])
self.assertIn("ctx.write_artifact", files["workspace-artifact-safety/SKILL.md"])
self.assertIn("live_skills[].input_schema", files["agent-quality-review/SKILL.md"])
backend = _with_builder_skills(StateBackend())
listing = backend.ls(BUILDER_SKILL_SOURCE)
paths = [entry["path"] for entry in (listing.entries or [])]
for skill_name in expected:
self.assertIn(f"{BUILDER_SKILL_SOURCE}{skill_name}/", paths)

30
tests/test_template_init.py
View File

@@ -6,10 +6,13 @@ import unittest
from agent_builder.tools import (
_BUILDER_STATE_FILE,
_BUILDER_INTERNAL_PREFIX,
_deployment_drift_error,
_files_from_tarball,
_parse_supporting_skill_files,
_replace_workspace_files,
_render_a2a_init_template,
_render_skill_md,
_tarball_workspace_dir,
)
@@ -25,6 +28,9 @@ class TemplateInitTests(unittest.TestCase):
self.assertIn('name = "research-agent"', files["agent.py"])
self.assertIn('description = "Research helper"', files["agent.py"])
self.assertIn("LLMProvisioning.CALLER_PROVIDED", files["agent.py"])
self.assertIn("WorkspaceAccess.dynamic", files["agent.py"])
self.assertIn("RUNTIME_SKILLS_ROOT", files["agent.py"])
self.assertIn("skills=skill_sources or None", files["agent.py"])
self.assertIn("ctx.llm", files["agent.py"])
self.assertIn("ctx.workspace_backend()", files["agent.py"])
self.assertIn("create_deep_agent", files["agent.py"])
@@ -85,6 +91,7 @@ class TemplateInitTests(unittest.TestCase):
s3 = _FakeS3({
prefix + "agent.py": b"print('ok')\n",
prefix + _BUILDER_STATE_FILE: b'{"repo_head_sha":"abc1234"}',
prefix + _BUILDER_INTERNAL_PREFIX + "skills/x/SKILL.md": b"hidden",
})
bundle = _tarball_workspace_dir(s3, "bucket", prefix)
@@ -92,6 +99,29 @@ class TemplateInitTests(unittest.TestCase):
with tarfile.open(fileobj=io.BytesIO(bundle), mode="r:gz") as tf:
self.assertEqual(tf.getnames(), ["agent.py"])
def test_render_skill_md_creates_valid_frontmatter(self) -> None:
text = _render_skill_md(
"market-research",
"Plan market research and synthesize findings.",
"Use subagents for separate market segments.",
)
self.assertIn("name: market-research", text)
self.assertIn('description: "Plan market research', text)
self.assertIn("# market-research", text)
def test_parse_supporting_skill_files_rejects_unsafe_paths(self) -> None:
with self.assertRaises(ValueError):
_parse_supporting_skill_files('{"../secrets.txt": "bad"}')
files = _parse_supporting_skill_files(
'{"references/schema.md": "schema", "scripts/check.py": "print(1)"}'
)
self.assertEqual(
sorted(files),
["references/schema.md", "scripts/check.py"],
)
def test_files_from_tarball_rejects_unsafe_paths(self) -> None:
bundle = _tarball({"../agent.py": "bad"})