a2a-source-edit: write agent.py

agent.py

@@ -1,25 +1,22 @@
-"""sits-between-parents-chaotic-19-aab317 agent.
+"""A2A agent for Brazilian preschool WhatsApp group triage."""
-
-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
-"""
 from __future__ import annotations
 
 import json
+import re
 from pathlib import Path
-from typing import Any
+from typing import Any, Literal
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 import a2a_pack as a2a
 from a2a_pack import (
     A2AAgent,
+    ConsumerSetup,
+    ConsumerSetupField,
     LLMProvisioning,
-    {{ auth_type }},
+    NoAuth,
     Pricing,
+    Resources,
     RunContext,
     WorkspaceAccess,
     WorkspaceMode,
@@ -28,131 +25,417 @@ from a2a_pack.context import LLMCreds
 
 
 class SitsBetweenParentsChaotic19Aab317Config(BaseModel):
-    pass
+    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 a compact tool-calling agent.
+You are the calm buffer between parents and a chaotic Brazilian preschool
+WhatsApp group.
 
-Use the text_stats tool when the user asks about text, counts, summaries,
+Your job: turn noisy Portuguese or English group messages into practical JSON
-or anything where exact length/word numbers would help. Mention tool results
+for a parent: checklist, calendar reminders, payment reminders, backpack alert
-briefly instead of dumping raw JSON.
+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, {{ auth_type }}]):
+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."
+    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 = {{ auth_type }}
+    auth_model = NoAuth
 
-    # 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.",
+        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,
     )
-    tools_used = ("deepagents", "langchain")
+    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")
 
-    @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 (
+            return TriageResult(
-                "LLM key required. Add an LLM credential in Settings > LLM "
+                status="setup_required",
-                "credentials before running this agent; for local --invoke "
+                answer=(
-                "runs set AGENT_LLM_KEY."
+                    "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},
         )
-        await ctx.emit_progress("deepagent finished")
+        raw = _last_message_text(state)
-        return _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[{{ auth_type }}],
+        ctx: RunContext[NoAuth],
         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
 
         @tool
-        def text_stats(text: str) -> str:
+        def validate_parent_plan_json(plan_json: str) -> str:
-            """Return exact word, character, and line counts for text."""
+            """Validate the JSON plan shape and flag missing parent-planning sections."""
-            words = [part for part in text.split() if part.strip()]
+            try:
-            return json.dumps(
+                data = json.loads(plan_json)
-                {
+            except Exception as exc:  # noqa: BLE001
-                    "characters": len(text),
+                return json.dumps({"valid": False, "error": f"invalid_json: {exc}"})
-                    "words": len(words),
+            required = [
-                    "lines": len(text.splitlines()) or 1,
+                "answer",
-                }
+                "school_tomorrow",
-            )
+                "checklist",
-
+                "calendar_reminders",
-        @wrap_model_call
+                "payment_reminders",
-        async def log_model_call(request: Any, handler: Any) -> Any:
+                "backpack_alerts",
-            messages = request.state.get("messages", [])
+                "morning_summaries",
-            print(
+            ]
-                "[middleware] model_call "
+            missing = [key for key in required if key not in data]
-                f"model={creds.model} source={creds.source} messages={len(messages)}"
+            return json.dumps({"valid": not missing, "missing": missing}, ensure_ascii=False)
-            )
-            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],
+            tools=[validate_parent_plan_json],
-            middleware=[log_model_call],
             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 _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/",)
+        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]:
-    """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 []
@@ -172,7 +455,6 @@ 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
@@ -187,3 +469,40 @@ def _last_message_text(state: dict[str, Any]) -> str:
                 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],
+    }