From 08377840a0f83b1992b51cf2ac9c26381bc7788d Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Mon, 13 Jul 2026 12:16:01 +0000 Subject: [PATCH] a2a-source-edit: write agent.py --- agent.py | 478 ++++++++++++++++++++++++++++++++++++++----------------- 1 file changed, 336 insertions(+), 142 deletions(-) diff --git a/agent.py b/agent.py index c8b852e..f68f8ce 100644 --- a/agent.py +++ b/agent.py @@ -1,189 +1,383 @@ -"""customer-integration-engineer agent. +"""Deterministic release validation agent for customer-integration-engineer. -Starter stack: - - DeepAgents for tool-calling orchestration - - Caller-provided LLM credentials via ctx.llm - - A tiny model-call middleware hook you can replace with tracing, - routing, rate limits, or policy checks +This agent validates a customer-integration-engineer release without editing the +validated source. It is intentionally local-logic only: no LLM credentials, +provider keys, code-editor-agent delegation, or cosmetic mutation paths. """ from __future__ import annotations import json -from pathlib import Path +import re from typing import Any +from urllib.parse import urlparse -from pydantic import BaseModel +from pydantic import BaseModel, Field import a2a_pack as a2a from a2a_pack import ( A2AAgent, - LLMProvisioning, - {{ auth_type }}, + NoAuth, Pricing, + Resources, RunContext, WorkspaceAccess, WorkspaceMode, ) -from a2a_pack.context import LLMCreds + + +EXPECTED_AGENT_NAME = "customer-integration-engineer" +EXPECTED_LIVE_VERSION = "0.1.5" +EXPECTED_PUBLIC = False +EXPECTED_WORKSPACE_MODES = {"read_only", "read_write_overlay"} +EXPECTED_SKILL_COUNT = 5 +MAX_ALLOWED_IMPROVEMENT_ITERATIONS = 0 +SECRET_KEY_RE = re.compile(r"(api[_-]?key|token|secret|password|authorization|bearer)", re.I) +SECRET_VALUE_RE = re.compile( + r"(?i)(sk-[A-Za-z0-9_-]{12,}|xox[baprs]-[A-Za-z0-9-]{12,}|gh[pousr]_[A-Za-z0-9_]{20,}|bearer\s+[A-Za-z0-9._-]{16,})" +) class CustomerIntegrationEngineerConfig(BaseModel): - pass + """Static release-validation policy.""" + + target_agent_name: str = EXPECTED_AGENT_NAME + target_version: str = EXPECTED_LIVE_VERSION + required_skill_count: int = EXPECTED_SKILL_COUNT -SYSTEM_PROMPT = """\ -You are a compact tool-calling agent. +class CardValidationInput(BaseModel): + """Bounded card details copied from the live Agent Card or registry view.""" -Use the text_stats tool when the user asks about text, counts, summaries, -or anything where exact length/word numbers would help. Mention tool results -briefly instead of dumping raw JSON. -""" - -RUNTIME_SKILLS_DIR = "customer-integration-engineer/.deepagents/skills/" -DEEPAGENTS_RECURSION_LIMIT = 500 + card: dict[str, Any] = Field(description="Live Agent Card JSON for the target release.") + expected_version: str = Field(default=EXPECTED_LIVE_VERSION, max_length=32) + expected_skill_count: int = Field(default=EXPECTED_SKILL_COUNT, ge=1, le=20) + require_private: bool = True -class CustomerIntegrationEngineer(A2AAgent[CustomerIntegrationEngineerConfig, {{ auth_type }}]): - name = "customer-integration-engineer" - description = "Validates the customer-integration-engineer release end to end, preserving existing source unless critical defects remain." +class SmokeCallInput(BaseModel): + """Structured result from an Agent Studio delegated live smoke invocation.""" + + skill_name: str = Field(min_length=1, max_length=80) + input_args: dict[str, Any] = Field(default_factory=dict) + result: dict[str, Any] = Field(default_factory=dict) + status_code: int = Field(default=200, ge=100, le=599) + used_workspace_delegation: bool = True + delegated_workspace_mode: str = Field(default="read_only", max_length=64) + + +class ReviewerFinding(BaseModel): + severity: str = Field(description="critical, warning, info, or pass", max_length=32) + area: str = Field(max_length=120) + message: str = Field(max_length=1000) + + +class CustomerIntegrationEngineer(A2AAgent[CustomerIntegrationEngineerConfig, NoAuth]): + name = EXPECTED_AGENT_NAME + description = ( + "Validates the customer-integration-engineer release end to end while " + "preserving source unless a critical defect remains." + ) version = "0.1.0" config_model = CustomerIntegrationEngineerConfig - auth_model = {{ auth_type }} + auth_model = NoAuth - # Hosted generated agents read the caller's saved LLM credential through - # ctx.llm. The platform may proxy that credential through LiteLLM, but agent - # code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY - # directly. - llm_provisioning = LLMProvisioning.PLATFORM pricing = Pricing( price_per_call_usd=0.0, - caller_pays_llm=True, - notes="Starter agent uses the caller's saved LLM credential via ctx.llm.", + caller_pays_llm=False, + notes="Deterministic validation checks only; no LLM credential required.", ) + resources = Resources(cpu="250m", memory="512Mi", max_runtime_seconds=300) workspace_access = WorkspaceAccess.dynamic( max_files=64, allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), require_reason=False, ) - tools_used = ("deepagents", "langchain") + tools_used = ("a2a-pack", "workspace-delegation-review") - @a2a.tool(description="Ask the starter DeepAgent to answer with tool calls when useful") - async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str: - creds = ctx.llm - await ctx.emit_progress(f"llm: {creds.model} via {creds.source}") - if not creds.api_key: - return ( - "LLM key required. Add an LLM credential in Settings > LLM " - "credentials before running this agent; for local --invoke " - "runs set AGENT_LLM_KEY." - ) - graph = self._build_deep_agent(ctx=ctx, creds=creds) - state = await graph.ainvoke( - {"messages": [{"role": "user", "content": prompt}]}, - config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT}, - ) - await ctx.emit_progress("deepagent finished") - return _last_message_text(state) - - def _build_deep_agent( + @a2a.tool( + description="Validate the live Agent Card version, privacy, five public skills, and workspace safety controls.", + timeout_seconds=60, + idempotent=True, + cost_class="cheap", + ) + async def validate_live_card( self, - *, - ctx: RunContext[{{ auth_type }}], - creds: LLMCreds, - ) -> Any: - # Lazy imports keep `a2a card` usable before local dependencies are - # installed. `a2a deploy` installs requirements.txt during the build. - from a2a_pack.deepagents import create_a2a_deep_agent - from langchain.agents.middleware import wrap_model_call - from langchain_core.tools import tool + ctx: RunContext[NoAuth], + payload: CardValidationInput, + ) -> dict[str, Any]: + await ctx.emit_progress("validating live card contract") + card = payload.card + skills = _skills(card) + workspace = _workspace_access(card) + runtime = _runtime(card) + checks = [ + _check("agent_name", card.get("name") == self.config.target_agent_name, f"expected {self.config.target_agent_name}"), + _check("version", str(card.get("version")) == payload.expected_version, f"expected live version {payload.expected_version}"), + _check("skill_count", len(skills) == payload.expected_skill_count, f"expected {payload.expected_skill_count} skills, got {len(skills)}"), + _check("private_exposure", _is_private(card) if payload.require_private else True, "agent must remain private"), + _check("workspace_enabled", bool(workspace.get("enabled")), "WorkspaceAccess.dynamic must be enabled"), + _check("workspace_modes", EXPECTED_WORKSPACE_MODES <= set(_mode_values(workspace.get("allowed_modes"))), "requires read_only and read_write_overlay"), + _check("no_llm_required", not _runtime_requires_caller_llm(runtime), "validator/smoke path must not require caller LLM"), + ] + return _result("validate_live_card", checks, extra={"skills": [_skill_name(s) for s in skills]}) - @tool - def text_stats(text: str) -> str: - """Return exact word, character, and line counts for text.""" - words = [part for part in text.split() if part.strip()] - return json.dumps( - { - "characters": len(text), - "words": len(words), - "lines": len(text.splitlines()) or 1, - } - ) - - @wrap_model_call - async def log_model_call(request: Any, handler: Any) -> Any: - messages = request.state.get("messages", []) - print( - "[middleware] model_call " - f"model={creds.model} source={creds.source} messages={len(messages)}" - ) - return await handler(request) - - backend = ctx.workspace_backend() - skill_sources = _seed_runtime_skills(backend, ctx) - # create_a2a_deep_agent resolves provider:model strings with - # langchain.init_chat_model from ctx.llm, preserving LiteLLM routing, - # provider-specific extra body, and runtime model overrides. - return create_a2a_deep_agent( - ctx, - creds=creds, - backend=backend, - skills=skill_sources or None, - tools=[text_stats], - middleware=[log_model_call], - system_prompt=SYSTEM_PROMPT, + @a2a.tool( + description="Confirm the primary public skill has bounded typed inputs and no credential-shaped parameters.", + timeout_seconds=60, + idempotent=True, + cost_class="cheap", + ) + async def validate_primary_skill_schema( + self, + ctx: RunContext[NoAuth], + payload: CardValidationInput, + primary_skill_name: str = "", + ) -> dict[str, Any]: + await ctx.emit_progress("validating primary skill schema") + skills = _skills(payload.card) + primary = _select_skill(skills, primary_skill_name) + schema = _input_schema(primary) if primary else {} + properties = schema.get("properties") if isinstance(schema, dict) else {} + required = schema.get("required") if isinstance(schema, dict) else [] + property_names = sorted(properties) if isinstance(properties, dict) else [] + checks = [ + _check("primary_skill_present", primary is not None, "primary skill must exist"), + _check("schema_is_object", isinstance(schema, dict) and schema.get("type") == "object", "input_schema must be a JSON object"), + _check("bounded_property_count", isinstance(properties, dict) and 1 <= len(properties) <= 12, "primary skill should expose 1-12 typed inputs"), + _check("additional_properties_false", schema.get("additionalProperties") is False, "schema should reject unknown inputs"), + _check("all_inputs_typed", _all_properties_typed(properties), "each input property must declare a type or structured schema"), + _check("no_credential_inputs", not any(SECRET_KEY_RE.search(name) for name in property_names), "public inputs must not request secrets"), + _check("required_is_bounded", isinstance(required, list) and len(required) <= len(property_names), "required list must be bounded"), + ] + return _result( + "validate_primary_skill_schema", + checks, + extra={"primary_skill": _skill_name(primary) if primary else None, "input_names": property_names}, ) + @a2a.tool( + description="Review source-change policy for critical-only edits, no code-editor-agent use, and version bump discipline.", + timeout_seconds=60, + idempotent=True, + cost_class="cheap", + ) + async def review_source_change_policy( + self, + ctx: RunContext[NoAuth], + mutated_source: bool, + critical_defect_remaining: bool, + used_code_editor_agent: bool, + improvement_iterations: int = 0, + previous_version: str = EXPECTED_LIVE_VERSION, + resulting_version: str = EXPECTED_LIVE_VERSION, + changed_files: list[str] | None = None, + ) -> dict[str, Any]: + await ctx.emit_progress("reviewing source mutation policy") + files = changed_files or [] + checks = [ + _check("no_code_editor_agent", not used_code_editor_agent, "code-editor-agent must not be used"), + _check("iteration_limit", 0 <= improvement_iterations <= MAX_ALLOWED_IMPROVEMENT_ITERATIONS, "at most 0 improvement iterations allowed"), + _check("critical_only_mutation", (not mutated_source) or critical_defect_remaining, "source may change only when a critical defect remains"), + _check("version_bumped_on_mutation", (not mutated_source) or previous_version != resulting_version, "version must change after real code-editor mutation"), + _check("no_cosmetic_files", not _looks_cosmetic_only(files), "avoid cosmetic-only source changes"), + ] + return _result("review_source_change_policy", checks, extra={"changed_files": files}) -def _runtime_skills_root(ctx: RunContext[Any]) -> str: - workspace = getattr(ctx, "_workspace", None) - prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ()) - if not prefixes: - outputs_prefix = getattr(workspace, "outputs_prefix", None) - prefixes = (outputs_prefix or "outputs/",) - prefix = str(prefixes[0]).strip("/") - return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}" + @a2a.tool( + description="Validate a live smoke-call result from Agent Studio workspace delegation and redact any secret-like values.", + timeout_seconds=60, + idempotent=True, + cost_class="cheap", + ) + async def validate_smoke_call( + self, + ctx: RunContext[NoAuth], + smoke: SmokeCallInput, + ) -> dict[str, Any]: + await ctx.emit_progress("validating smoke-call result") + secret_paths = _secret_paths(smoke.result) + checks = [ + _check("http_success", 200 <= smoke.status_code < 300, f"status_code={smoke.status_code}"), + _check("structured_result", isinstance(smoke.result, dict) and bool(smoke.result), "smoke result must be a non-empty JSON object"), + _check("workspace_delegated", smoke.used_workspace_delegation, "smoke must use Agent Studio workspace delegation"), + _check("delegation_mode_safe", smoke.delegated_workspace_mode in EXPECTED_WORKSPACE_MODES, "delegated mode must be read_only or read_write_overlay"), + _check("no_secrets_returned", not secret_paths, "result must not contain secret-shaped keys or values"), + ] + return _result( + "validate_smoke_call", + checks, + extra={"skill_name": smoke.skill_name, "secret_like_paths": secret_paths}, + ) + + @a2a.tool( + description="Combine validation outputs and reviewer findings into a final go/no-go release report.", + timeout_seconds=60, + idempotent=True, + cost_class="cheap", + ) + async def final_release_report( + self, + ctx: RunContext[NoAuth], + validation_results: list[dict[str, Any]], + reviewer_findings: list[ReviewerFinding] | None = None, + warnings_to_accept: list[str] | None = None, + ) -> dict[str, Any]: + await ctx.emit_progress("building final release report") + findings = reviewer_findings or [] + warnings = warnings_to_accept or [] + criticals = [f.model_dump() for f in findings if f.severity.lower() == "critical"] + failed_checks = [ + {"result_index": i, "check": check} + for i, result in enumerate(validation_results) + for check in result.get("checks", []) + if isinstance(check, dict) and not check.get("ok") + ] + residual_warnings = [f.model_dump() for f in findings if f.severity.lower() == "warning"] + all_warnings_accounted = len(residual_warnings) == 0 or bool(warnings) + checks = [ + _check("zero_critical_findings", not criticals, "reviewer must report zero critical findings"), + _check("all_validation_checks_pass", not failed_checks, "all prior validation checks must pass"), + _check("warnings_patched_or_listed", all_warnings_accounted, "warnings must be patched or listed as residual risks"), + ] + status = "pass" if all(c["ok"] for c in checks) else "fail" + return { + "ok": status == "pass", + "status": status, + "target_agent": self.config.target_agent_name, + "target_version": self.config.target_version, + "checks": checks, + "critical_findings": criticals, + "failed_checks": failed_checks, + "residual_risks": warnings or [f["message"] for f in residual_warnings], + "secrets_included": False, + } -def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]: - """Copy packaged DeepAgents skills into the invocation workspace. - - DeepAgents loads skills from its backend, while source-controlled - ``skills/`` folders live in the image. This bridge lets generated agents - ship reusable SKILL.md bundles without giving up durable A2A workspace - files. - """ - root = Path(__file__).parent / "skills" - if not root.exists(): - return [] - runtime_skills_root = _runtime_skills_root(ctx) - 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 [] +def _runtime(card: dict[str, Any]) -> dict[str, Any]: + value = card.get("runtime") + return value if isinstance(value, dict) else {} -def _last_message_text(state: dict[str, Any]) -> str: - messages = state.get("messages") or [] - if not messages: - return json.dumps(state, default=str) +def _workspace_access(card: dict[str, Any]) -> dict[str, Any]: + for key in ("workspace_access", "workspaceAccess", "workspace"): + value = card.get(key) + if isinstance(value, dict): + return value + runtime = _runtime(card) + for key in ("workspace_access", "workspaceAccess", "workspace"): + value = runtime.get(key) + if isinstance(value, dict): + return value + return {} - content = getattr(messages[-1], "content", None) - if isinstance(content, str): - return content - if isinstance(content, list): - parts: list[str] = [] - for item in content: - if isinstance(item, dict): - text = item.get("text") or item.get("content") - if text: - parts.append(str(text)) - elif item: - parts.append(str(item)) - return "\n".join(parts) if parts else json.dumps(content, default=str) - return str(content or messages[-1]) + +def _skills(card: dict[str, Any]) -> list[dict[str, Any]]: + value = card.get("skills") or card.get("tools") or [] + return [item for item in value if isinstance(item, dict)] if isinstance(value, list) else [] + + +def _skill_name(skill: dict[str, Any] | None) -> str: + if not isinstance(skill, dict): + return "" + return str(skill.get("name") or skill.get("id") or "") + + +def _select_skill(skills: list[dict[str, Any]], name: str) -> dict[str, Any] | None: + if name: + return next((skill for skill in skills if _skill_name(skill) == name), None) + return skills[0] if skills else None + + +def _input_schema(skill: dict[str, Any] | None) -> dict[str, Any]: + if not isinstance(skill, dict): + return {} + for key in ("input_schema", "inputSchema", "parameters"): + value = skill.get(key) + if isinstance(value, dict): + return value + return {} + + +def _mode_values(value: Any) -> list[str]: + if isinstance(value, list): + raw = value + elif isinstance(value, tuple): + raw = list(value) + elif isinstance(value, str): + raw = [value] + else: + raw = [] + return [str(item.get("value") if isinstance(item, dict) else item).lower() for item in raw] + + +def _is_private(card: dict[str, Any]) -> bool: + expose = card.get("expose") + if isinstance(expose, dict) and "public" in expose: + return expose.get("public") is False + visibility = str(card.get("visibility") or card.get("access") or "").lower() + if visibility in {"private", "unlisted"}: + return True + public = card.get("public") + return public is False if isinstance(public, bool) else True + + +def _runtime_requires_caller_llm(runtime: dict[str, Any]) -> bool: + provisioning = str(runtime.get("llm_provisioning") or runtime.get("llmProvisioning") or "").lower() + pricing = runtime.get("pricing") if isinstance(runtime.get("pricing"), dict) else {} + return provisioning in {"platform", "caller_provided", "platform_or_caller_provided"} and bool(pricing.get("caller_pays_llm")) + + +def _all_properties_typed(properties: Any) -> bool: + if not isinstance(properties, dict) or not properties: + return False + for schema in properties.values(): + if not isinstance(schema, dict): + return False + if not any(key in schema for key in ("type", "anyOf", "oneOf", "allOf", "$ref")): + return False + return True + + +def _looks_cosmetic_only(paths: list[str]) -> bool: + if not paths: + return False + non_cosmetic_exts = {".py", ".yaml", ".yml", ".toml", ".json", ".txt"} + cosmetic_markers = {"readme", "docs/", ".md"} + lowered = [p.lower() for p in paths] + return all(any(marker in p for marker in cosmetic_markers) for p in lowered) and not any( + any(p.endswith(ext) for ext in non_cosmetic_exts - {".txt"}) for p in lowered + ) + + +def _secret_paths(value: Any, prefix: str = "$") -> list[str]: + found: list[str] = [] + if isinstance(value, dict): + for key, item in value.items(): + path = f"{prefix}.{key}" + if SECRET_KEY_RE.search(str(key)): + found.append(path) + found.extend(_secret_paths(item, path)) + elif isinstance(value, list): + for i, item in enumerate(value[:100]): + found.extend(_secret_paths(item, f"{prefix}[{i}]")) + elif isinstance(value, str) and SECRET_VALUE_RE.search(value): + found.append(prefix) + return sorted(set(found)) + + +def _check(name: str, ok: bool, detail: str) -> dict[str, Any]: + return {"name": name, "ok": bool(ok), "detail": detail} + + +def _result(name: str, checks: list[dict[str, Any]], *, extra: dict[str, Any] | None = None) -> dict[str, Any]: + ok = all(item["ok"] for item in checks) + payload = {"ok": ok, "status": "pass" if ok else "fail", "checkset": name, "checks": checks, "secrets_included": False} + if extra: + payload.update(extra) + return json.loads(json.dumps(payload, default=str)) + + +def _is_safe_url(url: str) -> bool: + parsed = urlparse(url) + return parsed.scheme in {"https", "http"} and bool(parsed.netloc)