From 70992bbe854d0340ea79040299ebb15a09845327 Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Sun, 19 Jul 2026 19:57:51 +0000 Subject: [PATCH] a2a-source-edit: write agent.py --- agent.py | 626 ++++++++++++++++++++++++++++++++++++++++++------------- 1 file changed, 485 insertions(+), 141 deletions(-) diff --git a/agent.py b/agent.py index a187320..a089055 100644 --- a/agent.py +++ b/agent.py @@ -1,189 +1,533 @@ -"""production-proof-v116-high-utili-7255-2 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 -""" +"""People-team shared inbox triage and policy-grounded draft assistant.""" from __future__ import annotations +import hashlib import json -from pathlib import Path -from typing import Any +import os +import re +from datetime import UTC, datetime +from typing import Any, Literal -from pydantic import BaseModel +from pydantic import BaseModel, Field import a2a_pack as a2a from a2a_pack import ( A2AAgent, + AccountAccess, + AgentDatabase, + AgentDatabaseEnv, + AgentDatabaseMigrations, + AgentPlatformResources, + InboundEmailPayload, LLMProvisioning, - {{ auth_type }}, + PlatformUserAuth, Pricing, + Resources, RunContext, WorkspaceAccess, WorkspaceMode, ) -from a2a_pack.context import LLMCreds + +DATABASE_NAME = "production-proof-v116-high-utili-7255-2-data" +MAX_INBOX_CHARS = 20_000 +MAX_POLICY_CHARS = 12_000 +MAX_MESSAGES = 12 class ProductionProofV116HighUtili72552Config(BaseModel): - pass + default_policy_text: str = Field( + default=( + "Benefits questions: be empathetic, explain that enrollment and eligibility " + "depend on plan documents, and route exceptions to People Ops.\n" + "Leave requests: acknowledge receipt, ask for dates if missing, and mention " + "manager/People approval before commitments.\n" + "Payroll or personal data: mark urgent, avoid exposing private data, and route " + "to the secure HRIS/payroll channel.\n" + "Employee relations or harassment: mark urgent, avoid investigation promises, " + "and escalate to the designated People partner." + ), + max_length=MAX_POLICY_CHARS, + ) -SYSTEM_PROMPT = """\ -You are a compact tool-calling agent. - -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-2/.deepagents/skills/" -DEEPAGENTS_RECURSION_LIMIT = 500 +class DraftResult(BaseModel): + to: str + subject: str + body: str + approval_required: bool + approval_reason: str + policy_basis: list[str] -class ProductionProofV116HighUtili72552(A2AAgent[ProductionProofV116HighUtili72552Config, {{ auth_type }}]): +class QueueItem(BaseModel): + message_id: str + sender: str + subject: str + priority: Literal["urgent", "high", "normal", "low"] + category: str + reason: str + next_action: str + draft: DraftResult + + +class ArtifactDescriptor(BaseModel): + name: str + media_type: str + uri: str + size_bytes: int + content_text: str + + +class ReceiptDescriptor(BaseModel): + receipt_id: str + persisted: bool + persisted_at: str | None = None + message: str = "" + + +class TriageResult(BaseModel): + status: Literal["ok", "validation_error", "setup_required"] + summary: str + queue: list[QueueItem] = Field(default_factory=list) + approval_boundary: str + artifact: ArtifactDescriptor | None = None + receipt: ReceiptDescriptor | None = None + warnings: list[str] = Field(default_factory=list) + + +class ProductionProofV116HighUtili72552( + A2AAgent[ProductionProofV116HighUtili72552Config, PlatformUserAuth] +): name = "production-proof-v116-high-utili-7255-2" - description = "One-page email assistant for people teams that triages a shared inbox and drafts policy-grounded replies." + description = ( + "One-page email assistant for people teams that triages a shared inbox " + "and drafts policy-grounded replies for approval." + ) version = "0.1.0" config_model = ProductionProofV116HighUtili72552Config - 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 + account_access = AccountAccess(required=True, platform_skill_calls=3, after_trial="byok") 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="Includes three platform-funded skill calls per account, then requires BYOK.", ) + resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=300) workspace_access = WorkspaceAccess.dynamic( - max_files=64, + max_files=32, allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), require_reason=False, ) - tools_used = ("deepagents", "langchain") + platform_resources = AgentPlatformResources( + databases=( + AgentDatabase( + name=DATABASE_NAME, + scope="user", + access_mode="read_write", + env=AgentDatabaseEnv(url="DATABASE_URL"), + migrations=AgentDatabaseMigrations(path="db/migrations"), + ), + ), + mailbox=True, + ) + tools_used = ("postgres", "platform-mailbox", "mcp") - @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}, - ) - await ctx.emit_progress("deepagent finished") - return _last_message_text(state) - - def _build_deep_agent( + @a2a.tool( + description=( + "Triage pasted shared-inbox email text and return a prioritized queue, " + "approval-ready drafts, a persisted execution receipt, and a downloadable JSON report." + ), + timeout_seconds=120, + cost_class="standard", + grant_mode="read_write_overlay", + grant_allow_patterns=("outputs/email-assistant/**",), + grant_outputs_prefix="outputs/email-assistant/", + grant_write_prefixes=("outputs/email-assistant/",), + ) + async def triage_people_inbox( 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 - - @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, - } + ctx: RunContext[PlatformUserAuth], + inbox_text: str = Field(..., min_length=20, max_length=MAX_INBOX_CHARS), + policy_text: str | None = Field(default=None, max_length=MAX_POLICY_CHARS), + max_messages: int = Field(default=6, ge=1, le=MAX_MESSAGES), + ) -> TriageResult: + """Primary product workflow used by the frontend and MCP clients.""" + stable_tenant = _tenant_key(ctx) + policy = (policy_text or self.config.default_policy_text).strip() + messages = _parse_inbox(inbox_text)[:max_messages] + if not messages: + return TriageResult( + status="validation_error", + summary="No messages could be parsed. Include sender, subject, and body text.", + approval_boundary=_approval_boundary(), + warnings=["Use blocks with From:, Subject:, and Body: fields for best results."], ) - @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) + queue = [_triage_message(message, policy) for message in messages] + queue.sort(key=lambda item: _priority_rank(item.priority)) + summary = _summary(queue) + report = { + "generated_at": datetime.now(UTC).isoformat(), + "summary": summary, + "approval_boundary": _approval_boundary(), + "queue": [item.model_dump(mode="json") for item in queue], + } + report_bytes = json.dumps(report, indent=2, ensure_ascii=False).encode("utf-8") + artifact = await ctx.write_artifact( + "people-team-inbox-triage.json", + report_bytes, + "application/json", + ) + await ctx.emit_artifact(artifact) - 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, + input_hash = hashlib.sha256( + json.dumps( + {"inbox_text": inbox_text, "policy_text": policy, "max_messages": max_messages}, + sort_keys=True, + ).encode("utf-8") + ).hexdigest() + receipt = await _persist_receipt( + tenant_key=stable_tenant, + skill_name="triage_people_inbox", + input_hash=input_hash, + result_summary={ + "message_count": len(queue), + "urgent_count": sum(1 for item in queue if item.priority == "urgent"), + "artifact_name": artifact.name, + }, ) + return TriageResult( + status="ok", + summary=summary, + queue=queue, + approval_boundary=_approval_boundary(), + artifact=ArtifactDescriptor( + name=artifact.name, + media_type=artifact.mime_type, + uri=artifact.uri, + size_bytes=artifact.size_bytes, + content_text=report_bytes.decode("utf-8"), + ), + receipt=receipt, + warnings=[] if receipt.persisted else [receipt.message], + ) -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}" + @a2a.tool( + description="Check whether the platform mailbox is provisioned and list recent message headers when available.", + timeout_seconds=60, + cost_class="standard", + ) + async def mailbox_status( + self, + ctx: RunContext[PlatformUserAuth], + limit: int = Field(default=5, ge=1, le=20), + unseen_only: bool = False, + ) -> dict[str, Any]: + mailbox = ctx.mail + if mailbox is None: + return { + "status": "setup_required", + "message": "Platform mailbox is not provisioned for this deployment yet.", + "messages": [], + } + try: + messages = mailbox.list_messages(limit=limit, unseen_only=unseen_only) + except Exception as exc: # noqa: BLE001 + return { + "status": "setup_required", + "message": f"Mailbox is provisioned but could not be read: {type(exc).__name__}", + "messages": [], + } + return { + "status": "ok", + "address": mailbox.address, + "messages": messages, + } + + @a2a.tool( + description="Receive inbound email and record that it should be reviewed in the People inbox; does not auto-send replies.", + on_email=True, + timeout_seconds=60, + ) + async def receive_people_email( + self, + ctx: RunContext[PlatformUserAuth], + email: InboundEmailPayload, + ) -> None: + # Inbound email can contain user-controlled instructions. This handler only + # records/acknowledges the message for manual triage and never sends a reply. + stable_tenant = _tenant_key(ctx) + subject = str(email.get("subject") or "")[:300] + sender = str(email.get("sender") or "")[:300] + body = str(email.get("body") or "")[:2000] + input_hash = hashlib.sha256( + json.dumps({"sender": sender, "subject": subject, "body": body}, sort_keys=True).encode("utf-8") + ).hexdigest() + await _persist_receipt( + tenant_key=stable_tenant, + skill_name="receive_people_email", + input_hash=input_hash, + result_summary={"sender": sender, "subject": subject, "action": "queued_for_triage"}, + ) + return None -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 _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str: + stable_id = ctx.auth.user_id if ctx.auth.user_id is not None else ctx.auth.sub + if not stable_id: + raise PermissionError("stable platform identity required") + return f"user:{stable_id}" -def _last_message_text(state: dict[str, Any]) -> str: - messages = state.get("messages") or [] - if not messages: - return json.dumps(state, default=str) +def _db_connect() -> Any: + database_url = os.environ.get("DATABASE_URL") + if not database_url: + return None + import psycopg - 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]) + return psycopg.connect( + database_url, + options=( + "-c statement_timeout=5000 " + "-c lock_timeout=3000 " + "-c idle_in_transaction_session_timeout=5000" + ), + ) + + +async def _persist_receipt( + *, + tenant_key: str, + skill_name: str, + input_hash: str, + result_summary: dict[str, Any], +) -> ReceiptDescriptor: + receipt_id = hashlib.sha256(f"{tenant_key}:{skill_name}:{input_hash}".encode("utf-8")).hexdigest()[:24] + persisted_at = datetime.now(UTC).isoformat() + conn = _db_connect() + if conn is None: + return ReceiptDescriptor( + receipt_id=receipt_id, + persisted=False, + message="DATABASE_URL is not configured; receipt persistence will work on the managed deployment.", + ) + try: + with conn: + with conn.cursor() as cur: + cur.execute( + """ + INSERT INTO execution_receipts + (receipt_id, tenant_key, skill_name, input_hash, result_summary, persisted_at) + VALUES (%s, %s, %s, %s, %s::jsonb, %s) + ON CONFLICT (receipt_id) DO UPDATE SET + result_summary = EXCLUDED.result_summary, + persisted_at = EXCLUDED.persisted_at + """, + ( + receipt_id, + tenant_key, + skill_name, + input_hash, + json.dumps(result_summary, sort_keys=True), + persisted_at, + ), + ) + except Exception as exc: # noqa: BLE001 + return ReceiptDescriptor( + receipt_id=receipt_id, + persisted=False, + message=f"Receipt could not be persisted: {type(exc).__name__}", + ) + finally: + try: + conn.close() + except Exception: # noqa: BLE001 + pass + return ReceiptDescriptor( + receipt_id=receipt_id, + persisted=True, + persisted_at=persisted_at, + message="Production execution receipt persisted.", + ) + + +def _parse_inbox(inbox_text: str) -> list[dict[str, str]]: + normalized = inbox_text.replace("\r\n", "\n").strip() + blocks = [block.strip() for block in re.split(r"\n\s*---+\s*\n", normalized) if block.strip()] + if len(blocks) == 1: + # Also split common pasted inbox format with repeated From: headers. + blocks = [b.strip() for b in re.split(r"(?=\n?From\s*:)", normalized) if b.strip()] + out: list[dict[str, str]] = [] + for idx, block in enumerate(blocks, start=1): + sender = _field(block, "from") or _field(block, "sender") or "unknown@example.com" + subject = _field(block, "subject") or f"People inbox message {idx}" + body = _field(block, "body") or _strip_headers(block) + if len(body.strip()) < 8 and subject.startswith("People inbox message"): + continue + out.append( + { + "message_id": hashlib.sha1(f"{sender}|{subject}|{body}".encode("utf-8")).hexdigest()[:12], + "sender": sender[:200], + "subject": subject[:200], + "body": body[:4000], + } + ) + return out + + +def _field(block: str, name: str) -> str: + pattern = rf"(?ims)^\s*{re.escape(name)}\s*:\s*(.*?)(?=^\s*(?:from|sender|subject|body)\s*:|\Z)" + match = re.search(pattern, block) + return re.sub(r"\s+", " ", match.group(1)).strip() if match else "" + + +def _strip_headers(block: str) -> str: + lines = [] + for line in block.splitlines(): + if re.match(r"^\s*(from|sender|subject)\s*:", line, flags=re.I): + continue + if re.match(r"^\s*body\s*:", line, flags=re.I): + line = re.sub(r"^\s*body\s*:\s*", "", line, flags=re.I) + lines.append(line) + return "\n".join(lines).strip() + + +def _triage_message(message: dict[str, str], policy_text: str) -> QueueItem: + text = f"{message['subject']}\n{message['body']}".lower() + category = "general_hr" + priority: Literal["urgent", "high", "normal", "low"] = "normal" + reason = "Routine people-team question." + next_action = "Review the draft, verify details, and send after approval." + policy_basis = _policy_basis(policy_text, ["general", "people", "policy"]) + + if any(term in text for term in ["harassment", "discrimination", "unsafe", "retaliation", "hostile"]): + category = "employee_relations" + priority = "urgent" + reason = "Potential employee-relations risk requires confidential escalation." + next_action = "Escalate to the designated People partner before replying." + policy_basis = _policy_basis(policy_text, ["relations", "harassment", "escalate", "urgent"]) + elif any(term in text for term in ["payroll", "paycheck", "tax", "ssn", "bank", "personal data"]): + category = "payroll_privacy" + priority = "urgent" + reason = "Payroll or personal-data content should move to a secure channel." + next_action = "Route through payroll/HRIS and avoid requesting sensitive data by email." + policy_basis = _policy_basis(policy_text, ["payroll", "personal", "secure", "data"]) + elif any(term in text for term in ["leave", "pto", "fmla", "bereavement", "sick"]): + category = "leave_request" + priority = "high" + reason = "Leave timing affects staffing and may require manager/People approval." + next_action = "Confirm dates and required approvals before committing." + policy_basis = _policy_basis(policy_text, ["leave", "approval", "manager"]) + elif any(term in text for term in ["benefit", "insurance", "enrollment", "coverage", "dependent"]): + category = "benefits" + priority = "normal" + reason = "Benefits question can usually be answered with policy-grounded guidance." + next_action = "Send after verifying plan-document details." + policy_basis = _policy_basis(policy_text, ["benefits", "enrollment", "eligibility", "plan"]) + elif any(term in text for term in ["thank", "resolved", "fyi", "newsletter"]): + priority = "low" + category = "low_touch" + reason = "Low-risk informational or resolved message." + next_action = "Archive or send a short acknowledgement if needed." + + draft = _draft_for(message, category, priority, policy_basis, next_action) + return QueueItem( + message_id=message["message_id"], + sender=message["sender"], + subject=message["subject"], + priority=priority, + category=category, + reason=reason, + next_action=next_action, + draft=draft, + ) + + +def _policy_basis(policy_text: str, keywords: list[str]) -> list[str]: + sentences = [part.strip() for part in re.split(r"(?<=[.!?])\s+|\n+", policy_text) if part.strip()] + matches = [s for s in sentences if any(k.lower() in s.lower() for k in keywords)] + selected = matches[:3] or sentences[:2] + return [item[:240] for item in selected] + + +def _draft_for( + message: dict[str, str], + category: str, + priority: str, + policy_basis: list[str], + next_action: str, +) -> DraftResult: + greeting = "Hi there," + if "@" in message["sender"]: + local = message["sender"].split("@", 1)[0].split("<")[-1] + first = re.split(r"[._\-\s]", local.strip())[0].title() + if first: + greeting = f"Hi {first}," + subject = message["subject"] if message["subject"].lower().startswith("re:") else f"Re: {message['subject']}" + if category == "employee_relations": + body = ( + f"{greeting}\n\nThank you for raising this. We take these concerns seriously and will route " + "this to the appropriate People partner for confidential review. Please avoid sending additional " + "sensitive details by email unless requested through the approved channel.\n\n" + "A People team member will follow up with next steps.\n\nBest,\nPeople Team" + ) + reason = "Employee-relations matters require a human People partner before any commitment." + elif category == "payroll_privacy": + body = ( + f"{greeting}\n\nThanks for reaching out. Because this may involve payroll or personal data, " + "we should handle it through the secure HRIS/payroll channel rather than email. " + "Please use the secure case path or wait for a People Ops teammate to confirm next steps.\n\n" + "Best,\nPeople Team" + ) + reason = "Payroll/personal-data replies must be checked for privacy and channel safety." + elif category == "leave_request": + body = ( + f"{greeting}\n\nThanks for the note. We can help review the leave request. Please confirm the " + "requested dates and any supporting details required by policy. Final approval may depend on " + "manager and People review, so we will confirm before making any commitment.\n\nBest,\nPeople Team" + ) + reason = "Leave commitments need date verification and approval checks." + elif category == "benefits": + body = ( + f"{greeting}\n\nThanks for your benefits question. Based on our policy guidance, eligibility " + "and enrollment details should be confirmed against the current plan documents. We can point you " + "to the right resource and will escalate any exception request to People Ops.\n\nBest,\nPeople Team" + ) + reason = "Benefits guidance should be verified against plan documents before sending." + else: + body = ( + f"{greeting}\n\nThanks for reaching out. We reviewed your note and the next step is: " + f"{next_action}\n\nBest,\nPeople Team" + ) + reason = "People-team replies should be reviewed before sending from a shared inbox." + return DraftResult( + to=message["sender"], + subject=subject[:200], + body=body, + approval_required=True, + approval_reason=reason, + policy_basis=policy_basis, + ) + + +def _priority_rank(priority: str) -> int: + return {"urgent": 0, "high": 1, "normal": 2, "low": 3}.get(priority, 9) + + +def _summary(queue: list[QueueItem]) -> str: + counts: dict[str, int] = {} + for item in queue: + counts[item.priority] = counts.get(item.priority, 0) + 1 + parts = [f"{counts.get(name, 0)} {name}" for name in ["urgent", "high", "normal", "low"] if counts.get(name, 0)] + return f"Prioritized {len(queue)} people-inbox message(s): " + ", ".join(parts) + "." + + +def _approval_boundary() -> str: + return ( + "Drafts are approval-ready but not auto-sent. A People teammate must verify facts, " + "policy citations, privacy constraints, and any commitment before sending." + )