From 5f99e3f7223e8bf45a6516e6e3bbeb51b4732a45 Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Sun, 19 Jul 2026 19:57:13 +0000 Subject: [PATCH] a2a-source-edit: write agent.py --- agent.py | 600 +++++++++++++++++++++++++++++++++++++++++-------------- 1 file changed, 454 insertions(+), 146 deletions(-) diff --git a/agent.py b/agent.py index 15f1f99..450b304 100644 --- a/agent.py +++ b/agent.py @@ -1,189 +1,497 @@ -"""production-proof-v116-high-utili-7255-3 agent. +"""Production Proof v116: high-utility sales operations dashboard. -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 +A deterministic full-stack A2A product: platform auth, managed Postgres, +MCP-callable typed tools, downloadable artifacts, and durable execution receipts. """ from __future__ import annotations +import csv +import hashlib +import io import json -from pathlib import Path -from typing import Any +import os +import re +from contextlib import contextmanager +from datetime import UTC, datetime +from typing import Any, Literal -from pydantic import BaseModel +from pydantic import BaseModel, Field, field_validator import a2a_pack as a2a from a2a_pack import ( A2AAgent, - LLMProvisioning, - {{ auth_type }}, + AccountAccess, + AgentDatabase, + AgentDatabaseEnv, + AgentDatabaseMigrations, + AgentPlatformResources, + PlatformUserAuth, Pricing, + Resources, RunContext, - WorkspaceAccess, - WorkspaceMode, + State, ) -from a2a_pack.context import LLMCreds + +DB_NAME = "production-proof-v116-high-utili-7255-3-data" +MAX_UPDATES = 24 +MAX_TEXT_CHARS = 6000 class ProductionProofV116HighUtili72553Config(BaseModel): - pass + default_window_days: int = Field(default=30, ge=7, le=90) -SYSTEM_PROMPT = """\ -You are a compact tool-calling agent. +class SalesUpdate(BaseModel): + account: str = Field(min_length=1, max_length=80) + owner: str = Field(min_length=1, max_length=80) + stage: str = Field(min_length=1, max_length=40) + amount_usd: float = Field(ge=0, le=10_000_000) + forecast_category: Literal["commit", "best_case", "pipeline", "omitted"] = "pipeline" + status: Literal["on_track", "at_risk", "blocked", "slipped", "won", "lost"] = "on_track" + days_in_stage: int = Field(ge=0, le=365) + update: str = Field(default="", max_length=400) -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 = "production-proof-v116-high-utili-7255-3/.deepagents/skills/" -DEEPAGENTS_RECURSION_LIMIT = 500 + @field_validator("account", "owner", "stage", "update") + @classmethod + def _clean_text(cls, value: str) -> str: + return re.sub(r"\s+", " ", value.strip()) -class ProductionProofV116HighUtili72553(A2AAgent[ProductionProofV116HighUtili72553Config, {{ auth_type }}]): +class BuildDashboardRequest(BaseModel): + dashboard_name: str = Field(default="Weekly Sales Ops Pulse", min_length=3, max_length=80) + segment: str = Field(default="Mid-market", min_length=1, max_length=80) + period_label: str = Field(default="This week", min_length=1, max_length=60) + min_exception_amount_usd: float = Field(default=25_000, ge=0, le=10_000_000) + updates: list[SalesUpdate] = Field(default_factory=list, max_length=MAX_UPDATES) + notes: str = Field(default="", max_length=MAX_TEXT_CHARS) + + @field_validator("dashboard_name", "segment", "period_label", "notes") + @classmethod + def _clean(cls, value: str) -> str: + return re.sub(r"\s+", " ", value.strip()) + + +class DashboardArtifact(BaseModel): + name: str + media_type: str + uri: str + size_bytes: int + content: str + + +class DashboardResult(BaseModel): + status: Literal["ok", "setup_required", "error"] + dashboard_id: str + receipt_id: str + dashboard_data: list[dict[str, Any]] = Field(alias="dashboard-data") + filters: dict[str, Any] + trends: list[dict[str, Any]] + exceptions: list[dict[str, Any]] + history: list[dict[str, Any]] + summary: str + artifact: DashboardArtifact | None = None + + +class HistoryResult(BaseModel): + status: Literal["ok", "setup_required", "error"] + history: list[dict[str, Any]] + + +class ProductionProofV116HighUtili72553( + A2AAgent[ProductionProofV116HighUtili72553Config, PlatformUserAuth] +): name = "production-proof-v116-high-utili-7255-3" - description = "One-page B2B sales ops dashboard that turns recurring operational updates into filters, trends, exceptions, durable history, artifacts, and MCP-callable tools." + description = ( + "One-page B2B sales ops dashboard that turns recurring operational " + "updates into filters, trends, exceptions, durable history, receipts, " + "downloadable outputs, and MCP-callable tools." + ) version = "0.1.0" config_model = ProductionProofV116HighUtili72553Config - auth_model = {{ auth_type }} + auth_model = PlatformUserAuth - # 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 + state = State.DURABLE + resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=600) 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 dashboard logic. Account trial funds exactly 3 platform skill calls, then BYOK is required by the platform.", ) - 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}, + account_access = AccountAccess(required=True, platform_skill_calls=3, after_trial="byok") + tools_used = ("postgres", "mcp", "artifacts", "scheduled-snapshot") + capabilities = { + "mcp": {"tools": ["build_dashboard", "list_dashboard_history", "run_scheduled_snapshot"]}, + "schedules": {"tool": "run_scheduled_snapshot", "idempotent": True}, + "artifacts": {"tool": "build_dashboard", "media_type": "text/csv"}, + } + platform_resources = AgentPlatformResources( + databases=( + AgentDatabase( + name=DB_NAME, + 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) + ) - def _build_deep_agent( + @a2a.tool( + description="Build and persist a one-page B2B sales operations dashboard with filters, trends, exceptions, history, receipt, and CSV artifact.", + timeout_seconds=60, + cost_class="standard", + ) + async def build_dashboard( 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], + request: BuildDashboardRequest, + ) -> DashboardResult: + tenant = _tenant_key(ctx) + now = datetime.now(UTC) + normalized = _with_default_updates(request) + analysis = _analyze_updates(normalized) + dashboard_id = _stable_id("dash", tenant, normalized.model_dump(mode="json"), now.isoformat(timespec="seconds")) + receipt_id = _stable_id("rcpt", tenant, dashboard_id, ctx.task_id or "task") + csv_text = _dashboard_csv(analysis["rows"]) + artifact_ref = await ctx.write_artifact( + "sales-ops-dashboard.csv", + csv_text.encode("utf-8"), + "text/csv", + ) + await ctx.emit_artifact(artifact_ref) - @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( + history: list[dict[str, Any]] = [] + if not os.environ.get("DATABASE_URL"): + return DashboardResult( + status="setup_required", + dashboard_id=dashboard_id, + receipt_id=receipt_id, + **{ + "dashboard-data": analysis["rows"], + "filters": analysis["filters"], + "trends": analysis["trends"], + "exceptions": analysis["exceptions"], + "history": [], + "summary": "Managed Postgres is not configured yet; dashboard preview and artifact were generated without durable history.", + "artifact": _artifact_payload(artifact_ref, csv_text), + }, + ) + + payload = { + "request": normalized.model_dump(mode="json"), + "analysis": analysis, + "artifact": {"name": artifact_ref.name, "uri": artifact_ref.uri, "media_type": artifact_ref.mime_type}, + } + receipt_payload = { + "receipt_id": receipt_id, + "dashboard_id": dashboard_id, + "skill": "build_dashboard", + "status": "ok", + "input_hash": _hash_json(normalized.model_dump(mode="json")), + "artifact_name": artifact_ref.name, + "created_at": now.isoformat(), + } + try: + with _db() as conn: + with conn.transaction(): + _insert_dashboard(conn, tenant, dashboard_id, normalized, analysis, payload, now) + _insert_receipt(conn, tenant, receipt_id, dashboard_id, receipt_payload, now) + history = _fetch_history(conn, tenant, limit=6) + except Exception as exc: # noqa: BLE001 - return structured product error, never raw DB URL/secrets + await ctx.emit_error("Could not persist dashboard history", code="db_persist_failed") + return DashboardResult( + status="error", + dashboard_id=dashboard_id, + receipt_id=receipt_id, + **{ + "dashboard-data": analysis["rows"], + "filters": analysis["filters"], + "trends": analysis["trends"], + "exceptions": analysis["exceptions"], + "history": [], + "summary": f"Dashboard generated, but durable history could not be saved ({type(exc).__name__}).", + "artifact": _artifact_payload(artifact_ref, csv_text), + }, + ) + + return DashboardResult( + status="ok", + dashboard_id=dashboard_id, + receipt_id=receipt_id, + **{ + "dashboard-data": analysis["rows"], + "filters": analysis["filters"], + "trends": analysis["trends"], + "exceptions": analysis["exceptions"], + "history": history, + "summary": _summary(normalized, analysis), + "artifact": _artifact_payload(artifact_ref, csv_text), + }, + ) + + @a2a.tool( + description="List durable dashboard history for the signed-in user.", + timeout_seconds=30, + cost_class="cheap", + idempotent=True, + ) + async def list_dashboard_history( + self, + ctx: RunContext[PlatformUserAuth], + limit: int = 10, + ) -> HistoryResult: + tenant = _tenant_key(ctx) + bounded_limit = max(1, min(int(limit), 25)) + if not os.environ.get("DATABASE_URL"): + return HistoryResult(status="setup_required", history=[]) + try: + with _db() as conn: + rows = _fetch_history(conn, tenant, bounded_limit) + except Exception: # noqa: BLE001 + return HistoryResult(status="error", history=[]) + return HistoryResult(status="ok", history=rows) + + @a2a.tool( + description="Idempotent scheduled sales dashboard snapshot. Safe for recurring platform schedule/MCP calls using a stable run key.", + timeout_seconds=60, + cost_class="cheap", + idempotent=True, + ) + async def run_scheduled_snapshot( + self, + ctx: RunContext[PlatformUserAuth], + run_key: str, + segment: str = "Mid-market", + ) -> DashboardResult: + safe_key = re.sub(r"[^A-Za-z0-9_.:-]", "-", run_key.strip())[:80] or "manual" + fixture = BuildDashboardRequest( + dashboard_name=f"Scheduled Sales Ops Pulse {safe_key}", + segment=segment[:80] or "Mid-market", + period_label=safe_key, + updates=_default_updates(), + ) + return await self.build_dashboard(ctx, fixture) + + +def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str: + stable_id = ctx.auth.user_id or ctx.auth.sub + if not stable_id: + raise PermissionError("stable platform identity required") + return f"user:{stable_id}" + + +@contextmanager +def _db(): + import psycopg + + conn = psycopg.connect( + os.environ["DATABASE_URL"], + options="-c statement_timeout=5000 -c lock_timeout=3000 -c idle_in_transaction_session_timeout=10000", + ) + try: + yield conn + finally: + conn.close() + + +def _default_updates() -> list[SalesUpdate]: + return [ + SalesUpdate(account="Acme Manufacturing", owner="Priya", stage="Security review", amount_usd=84000, forecast_category="commit", status="blocked", days_in_stage=18, update="Waiting on vendor risk questionnaire."), + SalesUpdate(account="Northstar Health", owner="Mateo", stage="Proposal", amount_usd=52000, forecast_category="best_case", status="at_risk", days_in_stage=11, update="Economic buyer asked for ROI proof."), + SalesUpdate(account="Bluebird Logistics", owner="Priya", stage="Discovery", amount_usd=27000, forecast_category="pipeline", status="on_track", days_in_stage=4, update="Expansion pain confirmed."), + SalesUpdate(account="Summit Retail", owner="Nora", stage="Procurement", amount_usd=125000, forecast_category="commit", status="slipped", days_in_stage=31, update="Legal redlines moved close date."), + ] + + +def _with_default_updates(request: BuildDashboardRequest) -> BuildDashboardRequest: + if request.updates: + return request + data = request.model_dump() + data["updates"] = [item.model_dump() for item in _default_updates()] + return BuildDashboardRequest.model_validate(data) + + +def _analyze_updates(request: BuildDashboardRequest) -> dict[str, Any]: + updates = request.updates[:MAX_UPDATES] + total_pipeline = round(sum(item.amount_usd for item in updates), 2) + weighted = round(sum(item.amount_usd * _forecast_weight(item.forecast_category) for item in updates), 2) + owners = sorted({item.owner for item in updates}) + stages = sorted({item.stage for item in updates}) + statuses = sorted({item.status for item in updates}) + rows: list[dict[str, Any]] = [] + exceptions: list[dict[str, Any]] = [] + for item in updates: + risk_score = _risk_score(item) + row = { + "account": item.account, + "owner": item.owner, + "stage": item.stage, + "amount_usd": round(item.amount_usd, 2), + "forecast_category": item.forecast_category, + "status": item.status, + "days_in_stage": item.days_in_stage, + "risk_score": risk_score, + "update": item.update, + } + rows.append(row) + if item.status in {"blocked", "slipped", "at_risk"} and item.amount_usd >= request.min_exception_amount_usd: + exceptions.append( { - "characters": len(text), - "words": len(words), - "lines": len(text.splitlines()) or 1, + "account": item.account, + "owner": item.owner, + "reason": f"{item.status.replace('_', ' ')} deal worth ${item.amount_usd:,.0f}", + "recommended_action": _recommended_action(item), + "risk_score": risk_score, } ) + rows.sort(key=lambda row: (-row["risk_score"], -row["amount_usd"], row["account"])) + exceptions.sort(key=lambda row: (-row["risk_score"], row["account"])) + stage_totals: dict[str, float] = {} + owner_totals: dict[str, float] = {} + for item in updates: + stage_totals[item.stage] = stage_totals.get(item.stage, 0.0) + item.amount_usd + owner_totals[item.owner] = owner_totals.get(item.owner, 0.0) + item.amount_usd + trends = [ + {"metric": "pipeline_total", "label": "Pipeline", "value": total_pipeline}, + {"metric": "weighted_pipeline", "label": "Weighted pipeline", "value": weighted}, + {"metric": "exception_count", "label": "Exceptions", "value": len(exceptions)}, + {"metric": "stale_deals", "label": "Deals >14 days in stage", "value": sum(1 for item in updates if item.days_in_stage > 14)}, + ] + return { + "rows": rows, + "filters": { + "segment": request.segment, + "period_label": request.period_label, + "owners": owners, + "stages": stages, + "statuses": statuses, + "min_exception_amount_usd": request.min_exception_amount_usd, + }, + "trends": trends, + "exceptions": exceptions, + "stage_totals": {key: round(value, 2) for key, value in sorted(stage_totals.items())}, + "owner_totals": {key: round(value, 2) for key, value in sorted(owner_totals.items())}, + } - @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) - 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, +def _forecast_weight(category: str) -> float: + return {"commit": 0.9, "best_case": 0.55, "pipeline": 0.25, "omitted": 0.0}.get(category, 0.25) + + +def _risk_score(item: SalesUpdate) -> int: + score = {"blocked": 90, "slipped": 75, "at_risk": 65, "on_track": 25, "won": 5, "lost": 5}[item.status] + if item.days_in_stage > 21: + score += 15 + elif item.days_in_stage > 14: + score += 8 + if item.amount_usd >= 100_000: + score += 10 + return min(score, 100) + + +def _recommended_action(item: SalesUpdate) -> str: + if item.status == "blocked": + return "Assign an owner to remove the blocker within 24 hours." + if item.status == "slipped": + return "Confirm the new close plan and next mutual action." + return "Book an executive or value-proof follow-up before the next forecast call." + + +def _summary(request: BuildDashboardRequest, analysis: dict[str, Any]) -> str: + total = next(item["value"] for item in analysis["trends"] if item["metric"] == "pipeline_total") + return ( + f"{request.dashboard_name} has {len(analysis['rows'])} opportunities, " + f"${total:,.0f} in pipeline, and {len(analysis['exceptions'])} prioritized exceptions." + ) + + +def _dashboard_csv(rows: list[dict[str, Any]]) -> str: + output = io.StringIO() + fieldnames = ["account", "owner", "stage", "amount_usd", "forecast_category", "status", "days_in_stage", "risk_score", "update"] + writer = csv.DictWriter(output, fieldnames=fieldnames) + writer.writeheader() + for row in rows: + writer.writerow({key: row.get(key, "") for key in fieldnames}) + return output.getvalue() + + +def _artifact_payload(ref: Any, content: str) -> DashboardArtifact: + return DashboardArtifact( + name=ref.name, + media_type=ref.mime_type, + uri=ref.uri, + size_bytes=ref.size_bytes, + content=content[:120_000], + ) + + +def _insert_dashboard(conn: Any, tenant: str, dashboard_id: str, request: BuildDashboardRequest, analysis: dict[str, Any], payload: dict[str, Any], now: datetime) -> None: + conn.execute( + """ + INSERT INTO sales_dashboards + (tenant_key, dashboard_id, dashboard_name, segment, period_label, pipeline_total_usd, exception_count, payload, created_at) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s) + """, + ( + tenant, + dashboard_id, + request.dashboard_name, + request.segment, + request.period_label, + next(item["value"] for item in analysis["trends"] if item["metric"] == "pipeline_total"), + len(analysis["exceptions"]), + json.dumps(payload), + now, + ), + ) + + +def _insert_receipt(conn: Any, tenant: str, receipt_id: str, dashboard_id: str, payload: dict[str, Any], now: datetime) -> None: + conn.execute( + """ + INSERT INTO execution_receipts + (tenant_key, receipt_id, dashboard_id, skill_name, status, payload, created_at) + VALUES (%s, %s, %s, %s, %s, %s::jsonb, %s) + ON CONFLICT (tenant_key, receipt_id) DO UPDATE SET + status = EXCLUDED.status, + payload = EXCLUDED.payload + """, + (tenant, receipt_id, dashboard_id, "build_dashboard", "ok", json.dumps(payload), now), + ) + + +def _fetch_history(conn: Any, tenant: str, limit: int) -> list[dict[str, Any]]: + cur = conn.execute( + """ + SELECT dashboard_id, dashboard_name, segment, period_label, pipeline_total_usd, exception_count, created_at + FROM sales_dashboards + WHERE tenant_key = %s + ORDER BY created_at DESC + LIMIT %s + """, + (tenant, limit), + ) + rows = [] + for row in cur.fetchall(): + rows.append( + { + "dashboard_id": row[0], + "dashboard_name": row[1], + "segment": row[2], + "period_label": row[3], + "pipeline_total_usd": float(row[4]), + "exception_count": int(row[5]), + "created_at": row[6].isoformat() if hasattr(row[6], "isoformat") else str(row[6]), + } ) + return rows -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}" +def _stable_id(prefix: str, *parts: Any) -> str: + digest = hashlib.sha256("|".join(json.dumps(part, sort_keys=True, default=str) for part in parts).encode("utf-8")).hexdigest()[:16] + return f"{prefix}_{digest}" -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 _last_message_text(state: dict[str, Any]) -> str: - messages = state.get("messages") or [] - if not messages: - return json.dumps(state, default=str) - - 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 _hash_json(value: Any) -> str: + return hashlib.sha256(json.dumps(value, sort_keys=True, default=str).encode("utf-8")).hexdigest()