421 lines
15 KiB
Python
421 lines
15 KiB
Python
"""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"
|
|
DB_ENV_NAME = "DATABASE_URL"
|
|
|
|
|
|
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=DB_ENV_NAME),
|
|
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(DB_ENV_NAME):
|
|
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("managed receipt store is not configured 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(DB_ENV_NAME):
|
|
return ReceiptListResult(
|
|
status="setup_required",
|
|
receipts=[],
|
|
warning="Managed receipt storage 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[DB_ENV_NAME]
|
|
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'<rect width="{size}" height="{size}" rx="{size * 0.14:.1f}" fill="{bg}" opacity="0.18"/>'
|
|
if style != "outline"
|
|
else f'<rect x="{stroke}" y="{stroke}" width="{size - 2 * stroke}" height="{size - 2 * stroke}" rx="{size * 0.12:.1f}" fill="none" stroke="{primary}" stroke-width="{stroke}"/>'
|
|
)
|
|
glow = ""
|
|
if style == "neon":
|
|
glow = f'<filter id="glow"><feGaussianBlur stdDeviation="3" result="blur"/><feMerge><feMergeNode in="blur"/><feMergeNode in="SourceGraphic"/></feMerge></filter>'
|
|
|
|
body = _asset_body(asset_type, style, primary, accent, dark, stroke, px, pattern)
|
|
grid = ""
|
|
if style == "pixel":
|
|
grid = "".join(
|
|
f'<rect x="{px((pattern + i * 3) % 14 + 1)}" y="{px((pattern // 3 + i * 5) % 14 + 1)}" width="{px(1)}" height="{px(1)}" fill="{accent}" opacity="0.7"/>'
|
|
for i in range(5)
|
|
)
|
|
|
|
return (
|
|
f'<svg xmlns="http://www.w3.org/2000/svg" role="img" viewBox="0 0 {size} {size}" width="{size}" height="{size}">'
|
|
f'<title>{title}</title><defs>{glow}</defs>{bg_shape}'
|
|
f'<g filter="{"url(#glow)" if style == "neon" else "none"}">{grid}{body}</g>'
|
|
f'<text x="{size / 2:.1f}" y="{size - unit:.1f}" text-anchor="middle" font-family="monospace" font-size="{max(8, int(size / 13))}" fill="{dark}" opacity="0.78">{label}</text>'
|
|
"</svg>"
|
|
)
|
|
|
|
|
|
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'<circle cx="{px(8)}" cy="{px(4.2)}" r="{px(2.1)}" fill="{fill}" {common}/>'
|
|
f'<path d="M {px(5)} {px(7)} H {px(11)} L {px(12)} {px(13)} H {px(4)} Z" fill="{fill}" {common}/>'
|
|
f'<circle cx="{px(7.25)}" cy="{px(4)}" r="{px(.35)}" fill="{accent}"/><circle cx="{px(8.75)}" cy="{px(4)}" r="{px(.35)}" fill="{accent}"/>'
|
|
)
|
|
if asset_type == "enemy":
|
|
spikes = " ".join(f'{px(2 + i * 2)},{px(7 + ((pattern + i) % 3))}' for i in range(7))
|
|
return (
|
|
f'<polygon points="{px(8)},{px(2)} {px(14)},{px(8)} {px(11)},{px(14)} {px(5)},{px(14)} {px(2)},{px(8)}" fill="{fill}" {common}/>'
|
|
f'<polyline points="{spikes}" fill="none" stroke="{accent}" stroke-width="{stroke}"/>'
|
|
f'<circle cx="{px(6)}" cy="{px(8)}" r="{px(.55)}" fill="{dark}"/><circle cx="{px(10)}" cy="{px(8)}" r="{px(.55)}" fill="{dark}"/>'
|
|
)
|
|
if asset_type == "item":
|
|
return (
|
|
f'<path d="M {px(8)} {px(2)} L {px(13)} {px(7)} L {px(8)} {px(14)} L {px(3)} {px(7)} Z" fill="{fill}" {common}/>'
|
|
f'<path d="M {px(8)} {px(3.7)} L {px(10.8)} {px(7)} L {px(8)} {px(12)} L {px(5.2)} {px(7)} Z" fill="{accent}" opacity="0.85"/>'
|
|
)
|
|
if asset_type == "tile":
|
|
return "".join(
|
|
f'<rect x="{px(x)}" y="{px(y)}" width="{px(3)}" height="{px(3)}" fill="{primary if (x + y + pattern) % 2 else accent}" {common} opacity="0.9"/>'
|
|
for x in (2, 5, 8, 11)
|
|
for y in (2, 5, 8, 11)
|
|
)
|
|
if asset_type == "projectile":
|
|
return (
|
|
f'<path d="M {px(2)} {px(9)} C {px(6)} {px(3)}, {px(10)} {px(3)}, {px(14)} {px(7)} C {px(10)} {px(8)}, {px(7)} {px(10)}, {px(2)} {px(9)} Z" fill="{fill}" {common}/>'
|
|
f'<path d="M {px(3)} {px(9)} L {px(1)} {px(6)} M {px(3.4)} {px(9.5)} L {px(1)} {px(12)}" stroke="{accent}" stroke-width="{stroke}"/>'
|
|
)
|
|
return (
|
|
f'<rect x="{px(4)}" y="{px(4)}" width="{px(8)}" height="{px(8)}" rx="{px(1.4)}" fill="{fill}" {common}/>'
|
|
f'<path d="M {px(8)} {px(3)} V {px(13)} M {px(3)} {px(8)} H {px(13)}" stroke="{accent}" stroke-width="{stroke}"/>'
|
|
)
|