"""A2A agent that generates SVG video game assets and persists execution receipts."""
from __future__ import annotations
import hashlib
import html
import json
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, Field
import a2a_pack as a2a
from a2a_pack import (
A2AAgent,
AgentDatabase,
AgentDatabaseEnv,
AgentDatabaseMigrations,
AgentPlatformResources,
PlatformUserAuth,
PlatformUserAuthResolver,
Pricing,
Resources,
RunContext,
)
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):
default_size: int = Field(default=128, ge=32, le=512)
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 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 bounded SVG video game assets and records production execution receipts."
version = "0.1.0"
config_model = GeneratesSvgVideoGame9jg9vdF0f510Config
auth_model = PlatformUserAuth
auth_resolver = PlatformUserAuthResolver()
pricing = Pricing(
price_per_call_usd=0.0,
caller_pays_llm=False,
notes="Deterministic SVG generation; no LLM credential is required.",
)
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"),
),
)
)
tools_used = ("svg", "postgres")
@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[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"],
)
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(
{
"name": f"{_slugify(clean_theme)}-{asset_type}-{style}-{index + 1}.svg",
"media_type": "image/svg+xml",
"size": size,
"svg": svg,
}
)
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")
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,
)
@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 _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}"
@contextmanager
def _db_connection() -> Iterator[Any]:
import psycopg
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''
)