a2a-source-edit: write agent.py

This commit is contained in:
a2a-cloud
2026-07-18 16:55:12 +00:00
parent 965559cc3c
commit 3f3a1e1359

516
agent.py
View File

@@ -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,
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 = ("deepagents", "langchain")
)
tools_used = ("svg", "postgres")
@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."
@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",
)
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(
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(
{
"characters": len(text),
"words": len(words),
"lines": len(text.splitlines()) or 1,
"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,
}
)
@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)}"
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,
)
return await handler(request)
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.
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}"
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.
@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(
"""
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 []
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 _last_message_text(state: dict[str, Any]) -> str:
messages = state.get("messages") or []
if not messages:
return json.dumps(state, default=str)
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
]
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 _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}"/>'
)