513 lines
20 KiB
Python
513 lines
20 KiB
Python
"""A2A agent for Brazilian preschool WhatsApp group triage."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
import a2a_pack as a2a
|
|
from a2a_pack import (
|
|
A2AAgent,
|
|
ConsumerSetup,
|
|
ConsumerSetupField,
|
|
LLMProvisioning,
|
|
NoAuth,
|
|
Pricing,
|
|
Resources,
|
|
RunContext,
|
|
WorkspaceAccess,
|
|
WorkspaceMode,
|
|
)
|
|
from a2a_pack.context import LLMCreds
|
|
|
|
|
|
class SitsBetweenParentsChaotic19Aab317Config(BaseModel):
|
|
default_timezone: str = "America/Sao_Paulo"
|
|
default_backpack_alert_time: str = "20:00"
|
|
default_morning_summary_time: str = "06:45"
|
|
|
|
|
|
class ChecklistItem(BaseModel):
|
|
title: str
|
|
category: Literal[
|
|
"backpack",
|
|
"payment",
|
|
"calendar",
|
|
"form",
|
|
"medicine",
|
|
"snack_rotation",
|
|
"lost_item",
|
|
"other",
|
|
] = "other"
|
|
due_date: str | None = None
|
|
child_name: str | None = None
|
|
notes: str = ""
|
|
confidence: float = Field(default=0.8, ge=0.0, le=1.0)
|
|
|
|
|
|
class CalendarReminder(BaseModel):
|
|
title: str
|
|
date: str
|
|
time: str | None = None
|
|
timezone: str
|
|
all_day: bool = True
|
|
description: str = ""
|
|
kind: Literal[
|
|
"school_day",
|
|
"no_school",
|
|
"event",
|
|
"deadline",
|
|
"backpack_alert",
|
|
"morning_summary",
|
|
] = "event"
|
|
provider_action: Literal["proposed_only"] = "proposed_only"
|
|
|
|
|
|
class PaymentReminder(BaseModel):
|
|
label: str
|
|
amount: float | None = None
|
|
currency: str | None = "BRL"
|
|
due_date: str | None = None
|
|
status: Literal["paid", "unpaid", "unknown"] = "unknown"
|
|
approval_required: bool = True
|
|
provider_action: Literal["proposed_only"] = "proposed_only"
|
|
notes: str = ""
|
|
|
|
|
|
class IntegrationPreview(BaseModel):
|
|
integration: Literal["calendar", "payments"]
|
|
status: Literal["proposed", "setup_required", "not_applicable"]
|
|
action: str
|
|
summary: str
|
|
payload_preview: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class TriageResult(BaseModel):
|
|
status: Literal["ok", "setup_required", "error"] = "ok"
|
|
answer: str
|
|
school_tomorrow: Literal["yes", "no", "unknown"] = "unknown"
|
|
checklist: list[ChecklistItem] = Field(default_factory=list)
|
|
calendar_reminders: list[CalendarReminder] = Field(default_factory=list)
|
|
payment_reminders: list[PaymentReminder] = Field(default_factory=list)
|
|
backpack_alerts: list[CalendarReminder] = Field(default_factory=list)
|
|
morning_summaries: list[CalendarReminder] = Field(default_factory=list)
|
|
integration_previews: list[IntegrationPreview] = Field(default_factory=list)
|
|
clarifying_questions: list[str] = Field(default_factory=list)
|
|
warnings: list[str] = Field(default_factory=list)
|
|
raw_summary: str = ""
|
|
|
|
|
|
SYSTEM_PROMPT = """\
|
|
You are the calm buffer between parents and a chaotic Brazilian preschool
|
|
WhatsApp group.
|
|
|
|
Your job: turn noisy Portuguese or English group messages into practical JSON
|
|
for a parent: checklist, calendar reminders, payment reminders, backpack alert
|
|
the night before, morning summary, and direct answers such as "Does my child
|
|
have school tomorrow?" or "What do I need to send this week?".
|
|
|
|
Use the preschool-chaos-triage skill. Be careful with Brazilian preschool
|
|
context: no-class notices, pedagogical meetings, costume days, snack rotation,
|
|
birthday/Festa Junina contributions, medicine authorization, missing water
|
|
bottles, photo permission forms, and sudden requests for cardboard, EVA, glue,
|
|
or a family photo by tomorrow.
|
|
|
|
Never execute a payment or write to a real calendar. Only propose integration
|
|
actions. Never output secrets or credentials.
|
|
"""
|
|
|
|
RUNTIME_SKILLS_DIR = "sits-between-parents-chaotic-19-aab317/.deepagents/skills/"
|
|
DEEPAGENTS_RECURSION_LIMIT = 500
|
|
|
|
|
|
class SitsBetweenParentsChaotic19Aab317(A2AAgent[SitsBetweenParentsChaotic19Aab317Config, NoAuth]):
|
|
name = "sits-between-parents-chaotic-19-aab317"
|
|
description = (
|
|
"Turns chaotic Brazilian preschool WhatsApp messages into checklists, "
|
|
"calendar reminders, payment reminders, backpack alerts, and morning summaries."
|
|
)
|
|
version = "0.1.0"
|
|
|
|
config_model = SitsBetweenParentsChaotic19Aab317Config
|
|
auth_model = NoAuth
|
|
|
|
llm_provisioning = LLMProvisioning.PLATFORM
|
|
pricing = Pricing(
|
|
price_per_call_usd=0.0,
|
|
caller_pays_llm=True,
|
|
notes="Uses the caller's saved LLM credential via ctx.llm; payment and calendar actions are previews only.",
|
|
)
|
|
resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=600)
|
|
workspace_access = WorkspaceAccess.dynamic(
|
|
max_files=64,
|
|
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
|
|
require_reason=False,
|
|
)
|
|
consumer_setup = ConsumerSetup.from_fields(
|
|
ConsumerSetupField.config(
|
|
"CALENDAR_PROVIDER",
|
|
label="Calendar provider",
|
|
description="Optional calendar destination for proposed reminders, for example Google Calendar, Outlook, or iCloud.",
|
|
required=False,
|
|
input_type="select",
|
|
options=("Google Calendar", "Outlook", "iCloud", "Other"),
|
|
),
|
|
ConsumerSetupField.config(
|
|
"CALENDAR_ID",
|
|
label="Calendar name or ID",
|
|
description="Optional calendar identifier used only in proposed calendar payload previews.",
|
|
required=False,
|
|
),
|
|
ConsumerSetupField.config(
|
|
"PAYMENT_PROVIDER",
|
|
label="Payment provider",
|
|
description="Optional payment rail for proposed school contribution reminders, for example Pix or bank transfer.",
|
|
required=False,
|
|
input_type="select",
|
|
options=("Pix", "Bank transfer", "Cash", "Other"),
|
|
),
|
|
ConsumerSetupField.config(
|
|
"PAYMENT_RECIPIENT_LABEL",
|
|
label="Payment recipient label",
|
|
description="Optional non-secret label such as class treasurer or school office. Never put Pix keys or bank details here.",
|
|
required=False,
|
|
),
|
|
)
|
|
tools_used = ("deepagents", "langchain", "calendar-preview", "payments-preview")
|
|
|
|
@a2a.tool(
|
|
name="triage_preschool_messages",
|
|
description=(
|
|
"Read Brazilian preschool WhatsApp messages and return a structured checklist, "
|
|
"calendar reminders, payment reminders, backpack alerts, and morning summaries."
|
|
),
|
|
timeout_seconds=600,
|
|
cost_class="llm",
|
|
grant_mode="read_write_overlay",
|
|
grant_allow_patterns=("outputs/parent-chaos/**",),
|
|
grant_outputs_prefix="outputs/parent-chaos/",
|
|
grant_write_prefixes=("outputs/parent-chaos/",),
|
|
)
|
|
async def triage_preschool_messages(
|
|
self,
|
|
ctx: RunContext[NoAuth],
|
|
messages: list[str],
|
|
reference_date: str,
|
|
timezone: str = "America/Sao_Paulo",
|
|
child_name: str | None = None,
|
|
question: str = "",
|
|
known_school_days: list[str] | None = None,
|
|
already_paid_labels: list[str] | None = None,
|
|
create_integration_previews: bool = True,
|
|
) -> dict[str, Any]:
|
|
"""Interpret preschool group messages into parent-ready action items."""
|
|
if not messages or not any(str(message).strip() for message in messages):
|
|
return TriageResult(
|
|
status="error",
|
|
answer="Send at least one WhatsApp message to triage.",
|
|
warnings=["messages cannot be empty"],
|
|
).model_dump(mode="json")
|
|
|
|
creds = ctx.llm
|
|
if not creds.api_key:
|
|
return TriageResult(
|
|
status="setup_required",
|
|
answer=(
|
|
"LLM credentials are required. Add an LLM credential in A2A Cloud "
|
|
"Settings before running this agent."
|
|
),
|
|
warnings=["missing ctx.llm.api_key"],
|
|
).model_dump(mode="json")
|
|
|
|
tz = timezone or self.config.default_timezone
|
|
await ctx.emit_progress("Reading preschool WhatsApp messages")
|
|
graph = self._build_deep_agent(ctx=ctx, creds=creds)
|
|
prompt = self._build_prompt(
|
|
messages=messages[:80],
|
|
reference_date=reference_date,
|
|
timezone=tz,
|
|
child_name=child_name,
|
|
question=question,
|
|
known_school_days=known_school_days or [],
|
|
already_paid_labels=already_paid_labels or [],
|
|
create_integration_previews=create_integration_previews,
|
|
)
|
|
state = await graph.ainvoke(
|
|
{"messages": [{"role": "user", "content": prompt}]},
|
|
config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT},
|
|
)
|
|
raw = _last_message_text(state)
|
|
await ctx.emit_progress("Building structured parent summary")
|
|
parsed = _parse_json_object(raw)
|
|
if parsed is None:
|
|
return TriageResult(
|
|
status="error",
|
|
answer="I could not convert the messages into valid structured JSON.",
|
|
warnings=["LLM returned non-JSON output"],
|
|
raw_summary=raw[:4000],
|
|
).model_dump(mode="json")
|
|
|
|
try:
|
|
result = TriageResult.model_validate(parsed)
|
|
except Exception as exc: # noqa: BLE001
|
|
fallback = _coerce_minimal_result(parsed, raw, str(exc))
|
|
result = TriageResult.model_validate(fallback)
|
|
|
|
result.integration_previews.extend(
|
|
self._integration_previews(
|
|
result,
|
|
calendar_provider=str(ctx.consumer_config("CALENDAR_PROVIDER", "") or ""),
|
|
calendar_id=str(ctx.consumer_config("CALENDAR_ID", "") or ""),
|
|
payment_provider=str(ctx.consumer_config("PAYMENT_PROVIDER", "") or ""),
|
|
payment_recipient=str(ctx.consumer_config("PAYMENT_RECIPIENT_LABEL", "") or ""),
|
|
enabled=create_integration_previews,
|
|
)
|
|
)
|
|
return result.model_dump(mode="json")
|
|
|
|
@a2a.tool(
|
|
name="chat",
|
|
description="Chat with the preschool WhatsApp buffer and get a concise structured answer.",
|
|
timeout_seconds=600,
|
|
cost_class="llm",
|
|
grant_mode="read_write_overlay",
|
|
grant_allow_patterns=("outputs/parent-chaos/**",),
|
|
grant_outputs_prefix="outputs/parent-chaos/",
|
|
grant_write_prefixes=("outputs/parent-chaos/",),
|
|
)
|
|
async def chat(
|
|
self,
|
|
ctx: RunContext[NoAuth],
|
|
message: str,
|
|
reference_date: str,
|
|
timezone: str = "America/Sao_Paulo",
|
|
child_name: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Convenience chat endpoint for the packed UI."""
|
|
return await self.triage_preschool_messages(
|
|
ctx,
|
|
messages=[message],
|
|
reference_date=reference_date,
|
|
timezone=timezone,
|
|
child_name=child_name,
|
|
question=message,
|
|
known_school_days=[],
|
|
already_paid_labels=[],
|
|
create_integration_previews=True,
|
|
)
|
|
|
|
def _build_deep_agent(
|
|
self,
|
|
*,
|
|
ctx: RunContext[NoAuth],
|
|
creds: LLMCreds,
|
|
) -> Any:
|
|
from a2a_pack.deepagents import create_a2a_deep_agent
|
|
from langchain_core.tools import StructuredTool
|
|
|
|
validation_tool = StructuredTool.from_function(
|
|
func=_validate_parent_plan_json,
|
|
name="validate_parent_plan_json",
|
|
description="Validate the JSON plan shape and flag missing parent-planning sections.",
|
|
)
|
|
backend = ctx.workspace_backend()
|
|
skill_sources = _seed_runtime_skills(backend, ctx)
|
|
return create_a2a_deep_agent(
|
|
ctx,
|
|
creds=creds,
|
|
backend=backend,
|
|
skills=skill_sources or None,
|
|
tools=[validation_tool],
|
|
system_prompt=SYSTEM_PROMPT,
|
|
)
|
|
|
|
def _build_prompt(
|
|
self,
|
|
*,
|
|
messages: list[str],
|
|
reference_date: str,
|
|
timezone: str,
|
|
child_name: str | None,
|
|
question: str,
|
|
known_school_days: list[str],
|
|
already_paid_labels: list[str],
|
|
create_integration_previews: bool,
|
|
) -> str:
|
|
schema = TriageResult.model_json_schema()
|
|
payload = {
|
|
"messages": messages,
|
|
"reference_date": reference_date,
|
|
"timezone": timezone,
|
|
"child_name": child_name,
|
|
"question": question,
|
|
"known_school_days": known_school_days[:40],
|
|
"already_paid_labels": already_paid_labels[:80],
|
|
"default_backpack_alert_time": self.config.default_backpack_alert_time,
|
|
"default_morning_summary_time": self.config.default_morning_summary_time,
|
|
"create_integration_previews": create_integration_previews,
|
|
}
|
|
return (
|
|
"Interpret these preschool WhatsApp messages for a parent. "
|
|
"Return ONLY one valid JSON object matching this JSON Schema. "
|
|
"Do not wrap in markdown. Use Portuguese or English in the answer based on the input. "
|
|
"Payment/calendar integrations must be proposed_only, never executed.\n\n"
|
|
f"Input payload:\n{json.dumps(payload, ensure_ascii=False, indent=2)}\n\n"
|
|
f"Result JSON Schema:\n{json.dumps(schema, ensure_ascii=False, indent=2)}"
|
|
)
|
|
|
|
def _integration_previews(
|
|
self,
|
|
result: TriageResult,
|
|
*,
|
|
calendar_provider: str,
|
|
calendar_id: str,
|
|
payment_provider: str,
|
|
payment_recipient: str,
|
|
enabled: bool,
|
|
) -> list[IntegrationPreview]:
|
|
if not enabled:
|
|
return []
|
|
previews: list[IntegrationPreview] = []
|
|
calendar_items = result.calendar_reminders + result.backpack_alerts + result.morning_summaries
|
|
if calendar_items:
|
|
previews.append(
|
|
IntegrationPreview(
|
|
integration="calendar",
|
|
status="proposed" if calendar_provider else "setup_required",
|
|
action="create_calendar_reminders_preview",
|
|
summary=(
|
|
f"Propose {len(calendar_items)} calendar reminder(s)"
|
|
+ (f" for {calendar_provider}" if calendar_provider else "; configure CALENDAR_PROVIDER to target a calendar")
|
|
),
|
|
payload_preview={
|
|
"provider": calendar_provider or None,
|
|
"calendar_id": calendar_id or None,
|
|
"events": [item.model_dump(mode="json") for item in calendar_items[:20]],
|
|
},
|
|
)
|
|
)
|
|
payment_items = result.payment_reminders
|
|
if payment_items:
|
|
previews.append(
|
|
IntegrationPreview(
|
|
integration="payments",
|
|
status="proposed" if payment_provider else "setup_required",
|
|
action="create_payment_reminders_preview",
|
|
summary=(
|
|
f"Propose {len(payment_items)} payment reminder(s)"
|
|
+ (f" via {payment_provider}" if payment_provider else "; configure PAYMENT_PROVIDER for payment routing notes")
|
|
),
|
|
payload_preview={
|
|
"provider": payment_provider or None,
|
|
"recipient_label": payment_recipient or None,
|
|
"payments": [item.model_dump(mode="json") for item in payment_items[:20]],
|
|
"approval_required": True,
|
|
},
|
|
)
|
|
)
|
|
return previews
|
|
|
|
|
|
def _validate_parent_plan_json(plan_json: str) -> str:
|
|
try:
|
|
data = json.loads(plan_json)
|
|
except Exception as exc: # noqa: BLE001
|
|
return json.dumps({"valid": False, "error": f"invalid_json: {exc}"})
|
|
required = [
|
|
"answer",
|
|
"school_tomorrow",
|
|
"checklist",
|
|
"calendar_reminders",
|
|
"payment_reminders",
|
|
"backpack_alerts",
|
|
"morning_summaries",
|
|
]
|
|
missing = [key for key in required if key not in data]
|
|
return json.dumps({"valid": not missing, "missing": missing}, ensure_ascii=False)
|
|
|
|
|
|
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/parent-chaos/",)
|
|
prefix = str(prefixes[0]).strip("/")
|
|
return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}"
|
|
|
|
|
|
def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]:
|
|
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 _parse_json_object(text: str) -> dict[str, Any] | None:
|
|
cleaned = text.strip()
|
|
if cleaned.startswith("```"):
|
|
cleaned = re.sub(r"^```(?:json)?\s*", "", cleaned)
|
|
cleaned = re.sub(r"\s*```$", "", cleaned)
|
|
try:
|
|
value = json.loads(cleaned)
|
|
return value if isinstance(value, dict) else None
|
|
except Exception:
|
|
pass
|
|
match = re.search(r"\{.*\}", cleaned, flags=re.DOTALL)
|
|
if not match:
|
|
return None
|
|
try:
|
|
value = json.loads(match.group(0))
|
|
return value if isinstance(value, dict) else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _coerce_minimal_result(parsed: dict[str, Any], raw: str, validation_error: str) -> dict[str, Any]:
|
|
return {
|
|
"status": "ok",
|
|
"answer": str(parsed.get("answer") or parsed.get("summary") or raw[:1000]),
|
|
"school_tomorrow": parsed.get("school_tomorrow") if parsed.get("school_tomorrow") in {"yes", "no", "unknown"} else "unknown",
|
|
"checklist": parsed.get("checklist") if isinstance(parsed.get("checklist"), list) else [],
|
|
"calendar_reminders": parsed.get("calendar_reminders") if isinstance(parsed.get("calendar_reminders"), list) else [],
|
|
"payment_reminders": parsed.get("payment_reminders") if isinstance(parsed.get("payment_reminders"), list) else [],
|
|
"backpack_alerts": parsed.get("backpack_alerts") if isinstance(parsed.get("backpack_alerts"), list) else [],
|
|
"morning_summaries": parsed.get("morning_summaries") if isinstance(parsed.get("morning_summaries"), list) else [],
|
|
"integration_previews": parsed.get("integration_previews") if isinstance(parsed.get("integration_previews"), list) else [],
|
|
"clarifying_questions": parsed.get("clarifying_questions") if isinstance(parsed.get("clarifying_questions"), list) else [],
|
|
"warnings": [f"schema_validation_warning: {validation_error}"],
|
|
"raw_summary": raw[:4000],
|
|
}
|