a2a-source-edit: write agent.py
This commit is contained in:
a2a-cloud
600
agent.py
600
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()
|
||||
|
||||
Reference in New Issue
Block a user