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'"
+ )
+
+
+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''
+ )