From cb4318124496ab9bc4d59682b4da119b2329db5b Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Tue, 14 Jul 2026 21:38:31 +0000 Subject: [PATCH] a2a-source-edit: write agent.py --- agent.py | 463 ++++++++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 391 insertions(+), 72 deletions(-) diff --git a/agent.py b/agent.py index 591c83d..e4f6c87 100644 --- a/agent.py +++ b/agent.py @@ -1,25 +1,22 @@ -"""sits-between-parents-chaotic-19-aab317 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 for Brazilian preschool WhatsApp group triage.""" 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, -or anything where exact length/word numbers would help. Mention tool results -briefly instead of dumping raw JSON. +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, {{ 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 ( - "LLM key required. Add an LLM credential in Settings > LLM " - "credentials before running this agent; for local --invoke " - "runs set AGENT_LLM_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}, ) - await ctx.emit_progress("deepagent finished") - return _last_message_text(state) + 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[{{ 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: - """Return exact word, character, and line counts for text.""" - words = [part for part in text.split() if part.strip()] - return json.dumps( - { - "characters": len(text), - "words": len(words), - "lines": len(text.splitlines()) or 1, - } - ) - - @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) + def validate_parent_plan_json(plan_json: str) -> str: + """Validate the JSON plan shape and flag missing parent-planning sections.""" + 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) 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], + tools=[validate_parent_plan_json], 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], + }