From 3f3a1e13594a031dab757b5104f74ce704ad71c8 Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Sat, 18 Jul 2026 16:55:12 +0000 Subject: [PATCH] a2a-source-edit: write agent.py --- agent.py | 524 +++++++++++++++++++++++++++++++++++++++---------------- 1 file changed, 377 insertions(+), 147 deletions(-) diff --git a/agent.py b/agent.py index e7e042e..541db94 100644 --- a/agent.py +++ b/agent.py @@ -1,189 +1,419 @@ -"""generates-svg-video-game-9jg9vd-f0f510 agent. - -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 -""" +"""A2A agent that generates SVG video game assets and persists execution receipts.""" from __future__ import annotations +import hashlib +import html import json -from pathlib import Path -from typing import Any +import os +import re +import uuid +from contextlib import contextmanager +from datetime import datetime, timezone +from typing import Annotated, Any, Iterator, Literal -from pydantic import BaseModel +from pydantic import BaseModel, Field import a2a_pack as a2a from a2a_pack import ( A2AAgent, - LLMProvisioning, - {{ auth_type }}, + AgentDatabase, + AgentDatabaseEnv, + AgentDatabaseMigrations, + AgentPlatformResources, + PlatformUserAuth, + PlatformUserAuthResolver, Pricing, + Resources, RunContext, - WorkspaceAccess, - WorkspaceMode, ) -from a2a_pack.context import LLMCreds + +ASSET_TYPES = ("character", "enemy", "item", "tile", "icon", "projectile") +STYLES = ("pixel", "flat", "outline", "neon") +DB_OPTIONS = "-c statement_timeout=5000 -c lock_timeout=2000 -c idle_in_transaction_session_timeout=10000" class GeneratesSvgVideoGame9jg9vdF0f510Config(BaseModel): - pass + default_size: int = Field(default=128, ge=32, le=512) -SYSTEM_PROMPT = """\ -You are a compact tool-calling agent. - -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 = "generates-svg-video-game-9jg9vd-f0f510/.deepagents/skills/" -DEEPAGENTS_RECURSION_LIMIT = 500 +class SvgAssetResult(BaseModel): + status: Literal["ok", "validation_error"] + receipt_id: str | None = None + asset_id: str | None = None + asset_type: str + theme: str + style: str + size: int + variant_count: int + svg_assets: list[dict[str, str | int]] + receipt_persisted: bool + warnings: list[str] = Field(default_factory=list) -class GeneratesSvgVideoGame9jg9vdF0f510(A2AAgent[GeneratesSvgVideoGame9jg9vdF0f510Config, {{ auth_type }}]): +class ReceiptSummary(BaseModel): + receipt_id: str + asset_id: str + asset_type: str + theme: str + style: str + size: int + variant_count: int + created_at: str + + +class ReceiptListResult(BaseModel): + status: Literal["ok", "setup_required"] + receipts: list[ReceiptSummary] + warning: str | None = None + + +class GeneratesSvgVideoGame9jg9vdF0f510( + A2AAgent[GeneratesSvgVideoGame9jg9vdF0f510Config, PlatformUserAuth] +): name = "generates-svg-video-game-9jg9vd-f0f510" - description = "Generates SVG video game assets with a packed product UI and persisted execution receipts." + description = "Generates bounded SVG video game assets and records production execution receipts." version = "0.1.0" config_model = GeneratesSvgVideoGame9jg9vdF0f510Config - auth_model = {{ auth_type }} + auth_model = PlatformUserAuth + auth_resolver = PlatformUserAuthResolver() - # 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 SVG generation; no LLM credential is required.", ) - workspace_access = WorkspaceAccess.dynamic( - max_files=64, - allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), - require_reason=False, - ) - tools_used = ("deepagents", "langchain") - - @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}, + resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=120) + platform_resources = AgentPlatformResources( + databases=( + AgentDatabase( + name="generates-svg-video-game-9jg9vd-f0f510-data", + scope="user", + access_mode="read_write", + env=AgentDatabaseEnv(url="DATABASE_URL"), + migrations=AgentDatabaseMigrations(path="db/migrations"), + ), ) - await ctx.emit_progress("deepagent finished") - return _last_message_text(state) + ) + tools_used = ("svg", "postgres") - def _build_deep_agent( + @a2a.tool( + description="Generate one to four SVG video game assets from a bounded typed prompt and persist an execution receipt.", + timeout_seconds=120, + idempotent=False, + cost_class="media-light", + ) + async def generate_svg_asset( 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[PlatformUserAuth], + asset_type: Literal["character", "enemy", "item", "tile", "icon", "projectile"], + theme: Annotated[str, Field(min_length=2, max_length=80)], + palette: Annotated[list[str], Field(min_length=1, max_length=5)] = ("#7dd3fc", "#0f172a", "#facc15"), + size: Annotated[int, Field(ge=32, le=512)] = 128, + style: Literal["pixel", "flat", "outline", "neon"] = "pixel", + variant_count: Annotated[int, Field(ge=1, le=4)] = 1, + ) -> SvgAssetResult: + """Create SVG game assets and save a DB receipt when production DB is configured.""" + tenant = _tenant_key(ctx) + clean_theme = _clean_text(theme, max_len=80) + colors = _normalize_palette(palette) + if not clean_theme: + return SvgAssetResult( + status="validation_error", + asset_type=asset_type, + theme=theme, + style=style, + size=size, + variant_count=variant_count, + svg_assets=[], + receipt_persisted=False, + warnings=["theme must contain at least one visible character"], + ) - @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( + await ctx.emit_progress(f"generating {variant_count} {style} {asset_type} asset(s)") + seed_material = json.dumps( + { + "tenant": tenant, + "asset_type": asset_type, + "theme": clean_theme, + "palette": colors, + "size": size, + "style": style, + "variant_count": variant_count, + }, + sort_keys=True, + ) + seed = hashlib.sha256(seed_material.encode("utf-8")).hexdigest() + asset_id = f"asset_{seed[:16]}" + receipt_id = f"rcpt_{uuid.uuid4().hex}" + created_at = datetime.now(timezone.utc).isoformat() + + assets: list[dict[str, str | int]] = [] + for index in range(variant_count): + svg = _render_svg( + asset_type=asset_type, + theme=clean_theme, + palette=colors, + size=size, + style=style, + variant=index, + seed=seed, + ) + assets.append( { - "characters": len(text), - "words": len(words), - "lines": len(text.splitlines()) or 1, + "name": f"{_slugify(clean_theme)}-{asset_type}-{style}-{index + 1}.svg", + "media_type": "image/svg+xml", + "size": size, + "svg": svg, } ) - @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) + warnings: list[str] = [] + persisted = False + if os.environ.get("DATABASE_URL"): + try: + _persist_asset_and_receipt( + tenant_key=tenant, + receipt_id=receipt_id, + asset_id=asset_id, + request={ + "asset_type": asset_type, + "theme": clean_theme, + "palette": colors, + "size": size, + "style": style, + "variant_count": variant_count, + }, + result={"asset_count": len(assets), "asset_names": [a["name"] for a in assets]}, + created_at=created_at, + ) + persisted = True + except Exception as exc: # noqa: BLE001 + warnings.append(f"receipt persistence unavailable: {type(exc).__name__}") + else: + warnings.append("DATABASE_URL not configured; receipt persistence skipped in this environment") - 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, + await ctx.emit_progress("svg asset generation complete") + return SvgAssetResult( + status="ok", + receipt_id=receipt_id, + asset_id=asset_id, + asset_type=asset_type, + theme=clean_theme, + style=style, + size=size, + variant_count=variant_count, + svg_assets=assets, + receipt_persisted=persisted, + warnings=warnings, ) - -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="List recent persisted SVG generation receipts for the signed-in platform user.", + timeout_seconds=30, + idempotent=True, + cost_class="cheap", + ) + async def list_recent_receipts( + self, + ctx: RunContext[PlatformUserAuth], + limit: Annotated[int, Field(ge=1, le=25)] = 10, + ) -> ReceiptListResult: + tenant = _tenant_key(ctx) + if not os.environ.get("DATABASE_URL"): + return ReceiptListResult( + status="setup_required", + receipts=[], + warning="DATABASE_URL is not configured for this environment.", + ) + rows = _load_recent_receipts(tenant, limit) + return ReceiptListResult(status="ok", receipts=[ReceiptSummary(**row) for row in rows]) -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 _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str: + auth = ctx.auth + stable_id = auth.user_id if auth.user_id is not None else auth.sub + if not stable_id: + raise PermissionError("stable platform identity required") + return f"user:{stable_id}" -def _last_message_text(state: dict[str, Any]) -> str: - messages = state.get("messages") or [] - if not messages: - return json.dumps(state, default=str) +@contextmanager +def _db_connection() -> Iterator[Any]: + import psycopg - 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]) + database_url = os.environ["DATABASE_URL"] + with psycopg.connect(database_url, options=DB_OPTIONS) as conn: + yield conn + + +def _persist_asset_and_receipt( + *, + tenant_key: str, + receipt_id: str, + asset_id: str, + request: dict[str, Any], + result: dict[str, Any], + created_at: str, +) -> None: + with _db_connection() as conn: + with conn.transaction(): + conn.execute( + """ + INSERT INTO svg_asset_receipts + (tenant_key, receipt_id, asset_id, asset_type, theme, style, size, variant_count, request_payload, result_payload, created_at) + VALUES + (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s::jsonb, %s) + """, + ( + tenant_key, + receipt_id, + asset_id, + request["asset_type"], + request["theme"], + request["style"], + request["size"], + request["variant_count"], + json.dumps(request, sort_keys=True), + json.dumps(result, sort_keys=True), + created_at, + ), + ) + + +def _load_recent_receipts(tenant_key: str, limit: int) -> list[dict[str, Any]]: + with _db_connection() as conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT receipt_id, asset_id, asset_type, theme, style, size, variant_count, created_at + FROM svg_asset_receipts + WHERE tenant_key = %s + ORDER BY created_at DESC + LIMIT %s + """, + (tenant_key, limit), + ) + rows = cur.fetchall() + return [ + { + "receipt_id": row[0], + "asset_id": row[1], + "asset_type": row[2], + "theme": row[3], + "style": row[4], + "size": row[5], + "variant_count": row[6], + "created_at": row[7].isoformat() if hasattr(row[7], "isoformat") else str(row[7]), + } + for row in rows + ] + + +def _normalize_palette(palette: list[str] | tuple[str, ...]) -> list[str]: + out: list[str] = [] + for raw in palette: + value = str(raw).strip() + if not re.fullmatch(r"#[0-9a-fA-F]{6}", value): + raise ValueError(f"invalid hex color: {value!r}") + if value.lower() not in [item.lower() for item in out]: + out.append(value) + if not out: + raise ValueError("palette must include at least one #RRGGBB color") + return out[:5] + + +def _clean_text(value: str, *, max_len: int) -> str: + return re.sub(r"\s+", " ", str(value)).strip()[:max_len] + + +def _slugify(value: str) -> str: + slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") + return slug or "game-asset" + + +def _render_svg( + *, + asset_type: str, + theme: str, + palette: list[str], + size: int, + style: str, + variant: int, + seed: str, +) -> str: + bg = palette[(variant + 1) % len(palette)] + primary = palette[variant % len(palette)] + accent = palette[(variant + 2) % len(palette)] + dark = "#0f172a" + unit = size / 16 + stroke = max(2, int(size / 48)) + title = html.escape(f"{theme} {asset_type} {style} variant {variant + 1}") + label = html.escape(theme[:18]) + pattern = int(seed[variant * 2 : variant * 2 + 2], 16) + + def px(n: float) -> str: + return f"{n * unit:.2f}" + + bg_shape = ( + f'' + if style != "outline" + else f'' + ) + glow = "" + if style == "neon": + glow = f'' + + body = _asset_body(asset_type, style, primary, accent, dark, stroke, px, pattern) + grid = "" + if style == "pixel": + grid = "".join( + f'' + for i in range(5) + ) + + return ( + f'' + f'{title}{glow}{bg_shape}' + f'{grid}{body}' + f'{label}' + "" + ) + + +def _asset_body(asset_type: str, style: str, primary: str, accent: str, dark: str, stroke: int, px: Any, pattern: int) -> str: + fill = "none" if style == "outline" else primary + common = f'stroke="{dark}" stroke-width="{stroke}" stroke-linejoin="round" stroke-linecap="round"' + if asset_type == "character": + return ( + f'' + f'' + f'' + ) + if asset_type == "enemy": + spikes = " ".join(f'{px(2 + i * 2)},{px(7 + ((pattern + i) % 3))}' for i in range(7)) + return ( + f'' + f'' + f'' + ) + if asset_type == "item": + return ( + f'' + f'' + ) + if asset_type == "tile": + return "".join( + f'' + for x in (2, 5, 8, 11) + for y in (2, 5, 8, 11) + ) + if asset_type == "projectile": + return ( + f'' + f'' + ) + return ( + f'' + f'' + )