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

feat: initial agent-reviewer meta-agent

Browse Source 
A2A agent that audits another deployed agent's source pre-deploy.
Reads the target Gitea repo via a short-lived read-only token minted
through ctx.mint_gitea_token(), runs an inner DeepAgents graph backed
by GiteaBackend, and emits a typed ReviewReport.

Skill bundles:
- security-anti-patterns (credentials, eval, sandbox bypass, egress)
- a2apack-best-practices (class shape, decorators, types, timeouts)
- grant-scope-evaluation (declared vs actual workspace scope)

Tools:
- sandbox_a2a_card  : round-trips the project through microsandbox
- sandbox_ruff_check: objective static checks
- submit_review_report: typed final report

7 smoke tests pass; agent card loads cleanly via `a2a card`.
This commit is contained in:
robert
2026-05-28 09:59:40 -03:00
commit 64a601a85a
17 changed files with 1065 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
agent_reviewer/__init__.py Normal file
View File

@@ -0,0 +1,6 @@
"""agent-reviewer — pre-deploy audit of A2A agent source."""
from __future__ import annotations
from .builder import ReviewerContext, build_reviewer_graph
__all__ = ["ReviewerContext", "build_reviewer_graph"]

140
agent_reviewer/builder.py Normal file
View File

@@ -0,0 +1,140 @@
"""Inner deepagents graph for the reviewer.
Reads the target agent's source from Gitea (via :class:`GiteaBackend`),
consults the three skill bundles bundled with this agent, and emits a
typed :class:`ReviewReport` through the ``submit_review_report`` tool.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from a2a_pack import GiteaBackend
from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI
from .config import Settings, load_settings
from .tools import ReviewReport, ToolContext, build_tools
REVIEWER_SKILL_SOURCE = "/.agent-reviewer/skills/"
@dataclass(frozen=True)
class ReviewerContext:
agent_name: str
ref: str
gitea_token: str
gitea_owner: str
settings: Settings | None = None
llm_base_url: str | None = None
llm_api_key: str | None = None
llm_model: str | None = None
llm_temperature_mode: str | None = None
llm_temperature: float | None = None
llm_extra_body: dict[str, Any] | None = None
SYSTEM_PROMPT = """\
You are an A2A agent reviewer. You audit deployed agents before they ship.
Your job: read the source of one agent project in a Gitea repository,
identify security, correctness, ergonomics, and scope issues, then emit a
single structured review via ``submit_review_report``.
You have packaged DeepAgents skills loaded from ``/.agent-reviewer/skills/``:
- ``security-anti-patterns`` — credential leakage, unsafe execution,
sandbox bypass, egress concerns. Consult first.
- ``a2apack-best-practices`` — class structure, @skill decorator
contract, type hints, timeouts, error handling.
- ``grant-scope-evaluation`` — declared vs actual workspace scope,
write-prefix coverage, scope expansion correctness.
Workflow:
1. List files at the repo root with ``list_files`` (or use ``glob`` for
patterns). Read ``a2a.yaml``, ``agent.py``, and ``requirements.txt``
first. Then any imported modules under the package directory.
2. Apply each skill bundle to the source. Tabulate findings with
``severity`` (critical|warning|info) and ``category`` (security|
correctness|ergonomics|policy|scope). Be strict about CRITICAL,
conservative about INFO — false positives erode trust.
3. Call ``sandbox_a2a_card`` to verify the project actually loads. An
exit_code != 0 is itself a CRITICAL correctness finding; cite the
stderr line that broke things.
4. Call ``sandbox_ruff_check`` for objective static issues. Map S-prefix
security rules to security/warning, other rules to ergonomics/info.
5. Once you have your full list, call ``submit_review_report`` exactly
once with the JSON-encoded ReviewReport. ``ok=True`` iff zero CRITICAL.
Discipline:
- One review = one ``submit_review_report`` call. Multiple submissions
only on schema rejection; fix and resubmit.
- Every finding must cite ``file`` (and ``line`` when the static check
or AST scan gave you one).
- Don't invent findings to look thorough. An empty findings list with
``ok=True`` is the right answer for a clean agent.
- Don't deploy, don't push commits, don't modify files. You are
read-only against the target repo.
"""
def build_reviewer_graph(
ctx: ReviewerContext,
*,
completion_box: dict[str, ReviewReport],
) -> Any:
settings = ctx.settings or load_settings()
backend = GiteaBackend(
gitea_url=settings.gitea_internal,
owner=ctx.gitea_owner,
repo=ctx.agent_name,
ref=ctx.ref,
token=ctx.gitea_token,
)
async def _fetch_source() -> dict[str, bytes]:
tree = await backend._load_tree() # type: ignore[attr-defined]
files: dict[str, bytes] = {}
for entry in tree:
if entry.is_dir:
continue
try:
files[entry.path] = await backend._read_bytes(entry.path) # type: ignore[attr-defined]
except Exception: # noqa: BLE001
continue
return files
tools = build_tools(
ToolContext(
settings=settings,
agent_name=ctx.agent_name,
ref=ctx.ref,
fetch_source=_fetch_source,
completion_box=completion_box,
)
)
model_kwargs: dict[str, Any] = {
"model": ctx.llm_model or settings.litellm_model,
"base_url": ctx.llm_base_url or (settings.litellm_url + "/v1"),
"api_key": ctx.llm_api_key or settings.litellm_key,
"stream_usage": True,
}
if ctx.llm_temperature_mode != "omit":
model_kwargs["temperature"] = (
ctx.llm_temperature if ctx.llm_temperature is not None else 0.0
)
if ctx.llm_extra_body:
model_kwargs["extra_body"] = dict(ctx.llm_extra_body)
model = ChatOpenAI(**model_kwargs)
return create_deep_agent(
model=model,
tools=tools,
system_prompt=SYSTEM_PROMPT,
backend=backend,
skills=[REVIEWER_SKILL_SOURCE],
)

46
agent_reviewer/config.py Normal file
View File

@@ -0,0 +1,46 @@
"""Runtime config — env-driven."""
from __future__ import annotations
import os
from dataclasses import dataclass
@dataclass(frozen=True)
class Settings:
sandbox_url: str
sandbox_timeout_s: int
sandbox_token: str | None
gitea_internal: str
gitea_public: str
litellm_url: str
litellm_key: str
litellm_model: str
cp_url: str
image: str
def load_settings() -> Settings:
return Settings(
sandbox_url=(
os.environ.get("A2A_SANDBOX_URL")
or os.environ.get("SANDBOX_URL", "http://sandbox.sandbox.svc.cluster.local:8000")
),
sandbox_timeout_s=int(
os.environ.get("A2A_SANDBOX_TIMEOUT_S")
or os.environ.get("SANDBOX_TIMEOUT_S", "600")
),
sandbox_token=os.environ.get("A2A_SANDBOX_TOKEN") or os.environ.get("SANDBOX_TOKEN"),
gitea_internal=os.environ.get(
"A2A_GITEA_INTERNAL", "http://gitea-http.gitea.svc.cluster.local:3000"
),
gitea_public=os.environ.get("A2A_GITEA_PUBLIC", "http://gitea.a2acloud.io"),
litellm_url=os.environ.get(
"A2A_LITELLM_URL", "http://litellm.llm.svc.cluster.local:4000"
),
litellm_key=os.environ.get("A2A_LITELLM_KEY", ""),
litellm_model=os.environ.get("A2A_LITELLM_MODEL", "gpt-5.5"),
cp_url=os.environ.get(
"A2A_CP_URL", "http://control-plane.control-plane.svc.cluster.local"
),
image=os.environ.get("AGENT_REVIEWER_SANDBOX_IMAGE", "python:3.11-slim"),
)

75
agent_reviewer/skills/a2apack-best-practices/SKILL.md Normal file
View File

@@ -0,0 +1,75 @@
---
name: a2apack-best-practices
description: Shape of a well-formed A2A Pack agent. Use when reviewing agent.py for class structure, @skill decorators, type hints, error handling, idempotency, and timeouts.
---
# A2A Pack Best Practices
Use this skill when reviewing the *shape* of an agent — not its security
posture (that's `security-anti-patterns`) and not its grants (that's
`grant-scope-evaluation`).
## Class contract
* The class must inherit from `A2AAgent[ConfigT, AuthT]` with concrete
generic parameters. Missing parameters → `warning`.
* Class variables `name`, `description`, `version` must be set. Missing →
`warning`. `description` shorter than ~30 chars → `info`.
* `config_model` and `auth_model` must reference declared Pydantic
models (or `NoAuth`). A mismatched generic and class var → `warning`.
## @skill decorators
* Every public skill should declare `description`. Missing → `warning`.
* `timeout_seconds` should be set explicitly. Default is short; long-running
skills that omit it will time out in production. Missing on a skill that
does file work, LLM calls, or sandbox exec → `warning`.
* `stream=True` is required if the skill calls `ctx.emit_progress`,
`ctx.ask`, `ctx.collect`, or `ctx.request_scope`. Missing while the body
does emit → `critical`.
* `idempotent=True` matters for skills that can be safely retried. Skills
that write files, mutate state, or charge money → `info` if not set
explicitly (either direction is fine, just be intentional).
## Type hints
* Every skill parameter must have a concrete type hint (no `Any`, no missing
annotation). The Card's `input_schema` is built from these. Missing → `warning`.
* Return type should be `dict[str, Any]`, a concrete Pydantic model, or `str`.
Returning untyped values surprises callers → `info`.
## Error handling
* Skills should return structured error dicts (e.g. `{"error": "..."}`), not
raise unhandled exceptions for *expected* failure modes (bad input, missing
workspace, timeouts). Bare `raise` for invalid input → `info`.
* `try/except Exception` that swallows the error without logging or returning
→ `warning`.
## ctx.workspace / ctx.sandbox usage
* If the skill calls `ctx.workspace_backend()`, the class must declare
`workspace_access = WorkspaceAccess.dynamic(...)`. Mismatch → `critical`.
* Subprocesses that produce user-visible files must run through
`ctx.workspace_shell` or `ctx.workspace_python`, not raw
`asyncio.create_subprocess_exec`. Already covered in
`security-anti-patterns` but worth a second look here.
## Recursion limits
* If the skill invokes a DeepAgents graph via `ainvoke` or `astream_events`,
it must pass `config={"recursion_limit": 500}` to match agent-builder.
Missing → `warning`.
## Manifest (a2a.yaml)
* `name` must match the agent's `name` class var. Mismatch → `critical`.
* `entrypoint` must point at a real `module:Class`. Stub or broken → `critical`.
* `version` should be semantic (major.minor.patch). Bad shape → `info`.
## Severity rubric (specific to this skill)
* `critical` — breaks the runtime contract (stream/emit mismatch, manifest
doesn't import, name disagreement).
* `warning` — works today but degrades the developer or caller experience
(no timeout, no type hint, bare raise).
* `info` — style nudges.

74
agent_reviewer/skills/grant-scope-evaluation/SKILL.md Normal file
View File

@@ -0,0 +1,74 @@
---
name: grant-scope-evaluation
description: Evaluate whether a skill's declared workspace grants match what its code actually does. Use when reviewing @skill grant_* parameters against ctx.workspace, ctx.sandbox, and file-writing patterns in the skill body.
---
# Grant Scope Evaluation
A2A grants are the platform's principle-of-least-authority lever. Every
public skill declares the scope it intends to use; the platform mints a
matching grant; the runtime enforces it at the FUSE layer.
Your job is to flag mismatches between **declared** and **actual** scope.
## What the decorator declares
Look at the `@skill(...)` call. The relevant fields:
* `grant_mode` — `read_only`, `read_write_overlay`, or absent.
* `grant_allow_patterns` — fnmatch globs the skill may *read*.
* `grant_deny_patterns` — fnmatch globs explicitly denied (overrides allow).
* `grant_outputs_prefix` — single prefix the skill writes outputs under.
* `grant_write_prefixes` — multiple write prefixes (more flexible than
`outputs_prefix`).
* `grant_ttl_seconds` — how long the grant lives.
* `allow_scope_expansion` — whether `ctx.request_scope` may be called.
## What the skill body actually does
Read the skill body and tabulate:
1. **Reads** — every `ctx.workspace.read(...)` / `read_bytes`, and every
built-in DeepAgents `read_file` / `grep` / `glob`. Each read implies a
path or glob that must be matched by `grant_allow_patterns`.
2. **Writes** — every `ctx.workspace.write(...)`, `write_artifact`,
`ctx.workspace_shell` / `workspace_python` invocation that produces
files. Each write implies a path that must be under
`grant_outputs_prefix` or one of `grant_write_prefixes`.
3. **Scope expansion** — any call to `ctx.request_scope(...)`. If present
without `allow_scope_expansion=True` → `critical` (will raise at
runtime).
## Common findings
* **Over-broad declared scope** (`info`/`warning`): grants the agent
more than it uses, e.g. `grant_allow_patterns=("**",)` when the body
only reads `agents/{name}/**`. Tighten the declaration.
* **Under-declared writes** (`critical`): body writes to a path not under
any declared write prefix → FUSE returns `EACCES` and the skill fails.
* **Under-declared reads** (`warning`): body reads paths not in allow
patterns → reads return "not found" even when the file exists.
* **Missing `stream=True`** combined with `allow_scope_expansion=True`
(`critical`): scope expansion requires the SSE stream.
* **Long TTL on a fast skill** (`info`): `grant_ttl_seconds=3600` on a
10-second skill is sloppy. Match TTL to expected runtime + buffer.
## Path template variables
Decorators support `{name}` and similar template variables that the
runtime substitutes from skill args. When evaluating coverage:
* Treat `agents/{name}/**` as "any path matching that template after the
caller's `name` arg is substituted".
* Flag templates referencing args that don't exist on the skill (`critical`).
## Cross-skill consistency
If the agent has multiple skills writing to overlapping prefixes, they
should declare consistent patterns. Divergence is `warning`.
## Severity rubric
* `critical` — would fail at runtime (writes hit `EACCES`, scope expansion
without `stream`, template references missing arg).
* `warning` — works but mismatched (over-broad scope, divergent siblings).
* `info` — tighten-the-declaration suggestions.

58
agent_reviewer/skills/security-anti-patterns/SKILL.md Normal file
View File

@@ -0,0 +1,58 @@
---
name: security-anti-patterns
description: Common security mistakes in A2A agent source. Use when reviewing agent.py, a2a.yaml, or any imported module for credential leakage, unsafe code execution, network egress, or sandbox bypass.
---
# Security Anti-Patterns in A2A Agents
Review the agent's source for these high-confidence red flags. Each finding
must cite a file path and (when available) a line number. Severity guidance
is below — be strict about `critical`, conservative about `info`.
## Credential leakage (critical)
* Hard-coded API keys, tokens, passwords. Even in comments. Patterns to grep:
`sk-`, `gho_`, `AKIA`, `xoxb-`, `Bearer `, `password=`, `api_key="..."`.
* Reading provider keys directly: `os.environ["OPENAI_API_KEY"]`,
`A2A_LITELLM_KEY`, master keys. The agent **must** use `ctx.llm` instead.
* Echoing secrets into logs, `emit_progress`, error messages, or returned dicts.
* Writing secrets to the workspace as files.
## Unsafe execution (critical)
* `eval(...)`, `exec(...)`, `compile(...)` on caller-supplied content.
* `subprocess.run`/`Popen` with `shell=True` and a string built from caller input.
* `os.system(...)` with any caller input.
* `pickle.load`/`pickle.loads` from caller-supplied bytes.
* Running caller-supplied code *outside* `ctx.sandbox`. Any subprocess spawned
in the agent container is unsandboxed.
## Sandbox bypass (critical)
* Calling `asyncio.create_subprocess_exec` for work that should produce
durable files — those run in the agent container, not the sandbox. Use
`ctx.workspace_shell` or `ctx.workspace_python`.
* Writing files outside the agent's declared `grant_write_prefixes`. The
sandbox enforces this at the FUSE layer, but unscoped writes from the
agent container itself are still a smell.
## Network egress (warning, critical if combined with credential exfil)
* `httpx`/`requests`/`urllib` calls to arbitrary caller-supplied URLs without
declaring egress policy. Check `runtime = AgentRuntime(egress=...)`.
* Webhooks or callbacks that send caller data to a third party not declared
in the agent card.
## Resource shape mismatches (warning)
* Long-running shell commands without `timeout` set.
* Unbounded loops over caller-supplied collections.
* Reading large files entirely into memory when streaming would do.
## Severity rubric
* `critical` — exploitable today, or directly violates the platform contract.
* `warning` — would fail a careful review; ship-blocker depending on context.
* `info` — style or hardening suggestion; does not block deploy.
When in doubt, prefer `warning` over `critical`. Reviewers that cry wolf get
ignored. Reviewers that miss credential leakage get fired.

190
agent_reviewer/tools.py Normal file
View File

@@ -0,0 +1,190 @@
"""Tools exposed to the inner DeepAgents graph.
The reviewer relies heavily on the LLM reading source through the
GiteaBackend (built-in DeepAgents `read_file`, `grep`, `glob`) plus the
skill bundles for judgment. Tools here cover what the model cannot do
on its own: round-trip the project through a sandbox to verify ``a2a
card`` succeeds, run ``ruff`` for objective static issues, and submit a
typed final report.
"""
from __future__ import annotations
import base64
import io
import json
import tarfile
import textwrap
from dataclasses import dataclass
from typing import Any, Literal
import httpx
from langchain_core.tools import tool
from pydantic import BaseModel, Field
from .config import Settings
_FINDING_LITERAL = Literal["critical", "warning", "info"]
_CATEGORY_LITERAL = Literal["security", "correctness", "ergonomics", "policy", "scope"]
class Finding(BaseModel):
severity: _FINDING_LITERAL
category: _CATEGORY_LITERAL
message: str
file: str | None = None
line: int | None = None
suggestion: str | None = None
class ReviewReport(BaseModel):
ok: bool = Field(
..., description="True iff there are zero critical findings."
)
agent_name: str
ref: str
summary: str
findings: list[Finding] = Field(default_factory=list)
@dataclass(frozen=True)
class ToolContext:
settings: Settings
agent_name: str
ref: str
fetch_source: Any # async callable: () -> dict[str, bytes]
completion_box: dict[str, ReviewReport]
def build_tools(ctx: ToolContext) -> list[Any]:
"""Construct LangChain tools bound to ``ctx``."""
sandbox_url = ctx.settings.sandbox_url.rstrip("/")
sandbox_timeout = ctx.settings.sandbox_timeout_s
sandbox_headers: dict[str, str] = {}
if ctx.settings.sandbox_token:
sandbox_headers["Authorization"] = f"Bearer {ctx.settings.sandbox_token}"
async def _tarball() -> bytes:
files: dict[str, bytes] = await ctx.fetch_source()
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz") as tf:
for path, data in files.items():
info = tarfile.TarInfo(name=path)
info.size = len(data)
tf.addfile(info, io.BytesIO(data))
return buf.getvalue()
async def _run_in_sandbox(script: str) -> dict[str, Any]:
tar_b64 = base64.b64encode(await _tarball()).decode()
payload = {
"command": script,
"image": ctx.settings.image,
"stdin_b64": tar_b64,
"timeout_seconds": sandbox_timeout,
}
async with httpx.AsyncClient(timeout=sandbox_timeout + 30) as client:
resp = await client.post(
f"{sandbox_url}/v1/run_shell",
json=payload,
headers=sandbox_headers,
)
if resp.status_code >= 400:
return {"error": f"sandbox {resp.status_code}: {resp.text[:300]}"}
return resp.json()
@tool
async def sandbox_a2a_card() -> dict[str, Any]:
"""Round-trip the agent project through a sandbox.
Extracts the Gitea source into ``/workspace``, installs requirements,
and runs ``a2a card`` to verify the manifest + class load cleanly.
Returns ``{"ok": True, "card": {...}}`` on success, or
``{"ok": False, "exit_code": int, "stdout": str, "stderr": str}`` on
any failure. Treat exit_code != 0 as a CRITICAL finding.
"""
script = textwrap.dedent(
"""
set -e
mkdir -p /workspace/project
cd /workspace/project
tar -xzf - 2>/dev/null
pip install --quiet -r requirements.txt 2>&1 | tail -5
a2a card
"""
).strip()
out = await _run_in_sandbox(script)
if out.get("exit_code", 0) != 0:
return {
"ok": False,
"exit_code": out.get("exit_code"),
"stdout": (out.get("stdout") or "")[-2000:],
"stderr": (out.get("stderr") or "")[-2000:],
}
try:
card = json.loads(out.get("stdout") or "{}")
except json.JSONDecodeError:
return {
"ok": False,
"exit_code": 0,
"error": "a2a card stdout was not valid JSON",
"stdout": (out.get("stdout") or "")[-2000:],
}
return {"ok": True, "card": card}
@tool
async def sandbox_ruff_check() -> dict[str, Any]:
"""Run ``ruff check`` on the agent source in a sandbox.
Returns ``{"exit_code": int, "findings": [...]}`` where each entry
carries ``code``, ``message``, ``file``, ``line``. Treat any finding
as at least an INFO; security-related codes (S-prefix) should be
WARNING or CRITICAL depending on the rule.
"""
script = textwrap.dedent(
"""
set -e
mkdir -p /workspace/project
cd /workspace/project
tar -xzf - 2>/dev/null
pip install --quiet ruff 2>&1 | tail -3
ruff check --output-format=json . || true
"""
).strip()
out = await _run_in_sandbox(script)
stdout = out.get("stdout") or ""
try:
findings = json.loads(stdout) if stdout.strip() else []
except json.JSONDecodeError:
findings = []
return {
"exit_code": out.get("exit_code", 0),
"stderr": (out.get("stderr") or "")[-1000:],
"findings": findings[:200],
}
@tool
async def submit_review_report(report_json: str) -> dict[str, Any]:
"""Submit the final structured review report. Call exactly once.
``report_json`` must parse to ``{ok, agent_name, ref, summary,
findings: [{severity, category, message, file?, line?, suggestion?}]}``.
Use lowercase severities (critical|warning|info) and categories
(security|correctness|ergonomics|policy|scope).
Returns ``{"accepted": True}`` on success or ``{"accepted": False,
"error": "..."}`` on schema failure — fix the JSON and call again.
"""
try:
raw = json.loads(report_json)
report = ReviewReport.model_validate(raw)
except Exception as exc: # noqa: BLE001
return {"accepted": False, "error": f"{type(exc).__name__}: {exc}"}
report = report.model_copy(update={
"agent_name": ctx.agent_name,
"ref": ctx.ref,
})
ctx.completion_box["report"] = report
return {"accepted": True}
return [sandbox_a2a_card, sandbox_ruff_check, submit_review_report]