""" Launch Desk — Function Tools ============================= Four @function_tool decorated tools for the Launch Planner agent. Each does simple but realistic logic without external API calls. """ from __future__ import annotations import re from agents import function_tool # --------------------------------------------------------------------------- # 1. extract_tasks # --------------------------------------------------------------------------- @function_tool def extract_tasks(brief: str) -> list[dict]: """Parse a product brief and extract a list of launch tasks. Each task includes a name, priority (high/medium/low), owner role, and a relative deadline expressed as days before launch. Args: brief: Free-form product brief text. Returns: A list of task dicts: {task, priority, owner, deadline} """ if not brief or not brief.strip(): return [ {"task": "Define launch goals and metrics", "priority": "high", "owner": "Product", "deadline": "T-30 days"}, ] text = brief.lower() # Heuristic keyword → task mapping. We scan the brief for domain keywords # and build a prioritised task list from whatever matches. task_templates: list[dict] = [ {"keywords": ["api", "backend", "server", "database", "infra"], "task": "Freeze code and tag release candidate", "priority": "high", "owner": "Engineering", "deadline": "T-14 days"}, {"keywords": ["qa", "test", "bug", "regression", "testing"], "task": "Complete QA regression and smoke testing", "priority": "high", "owner": "QA", "deadline": "T-7 days"}, {"keywords": ["doc", "documentation", "guide", "readme", "manual"], "task": "Write user-facing documentation and API guides", "priority": "medium", "owner": "Technical Writing", "deadline": "T-10 days"}, {"keywords": ["market", "marketing", "campaign", "ad", "social", "blog", "press"], "task": "Prepare marketing campaign assets and schedule posts", "priority": "high", "owner": "Marketing", "deadline": "T-5 days"}, {"keywords": ["support", "customer", "help", "ticket", "oncall"], "task": "Staff support team and prepare runbooks", "priority": "medium", "owner": "Customer Success", "deadline": "T-3 days"}, {"keywords": ["security", "compliance", "audit", "gdpr", "soc"], "task": "Complete security review and compliance sign-off", "priority": "high", "owner": "Security", "deadline": "T-21 days"}, {"keywords": ["pricing", "billing", "plan", "tier", "subscription"], "task": "Finalise pricing tiers and billing configuration", "priority": "medium", "owner": "Finance", "deadline": "T-12 days"}, {"keywords": ["analytics", "tracking", "metric", "dashboard", "kpi"], "task": "Set up analytics dashboards and success metrics", "priority": "low", "owner": "Data", "deadline": "T-8 days"}, {"keywords": ["design", "ui", "ux", "figma", "mockup", "brand"], "task": "Finalise UI polish and brand assets", "priority": "medium", "owner": "Design", "deadline": "T-10 days"}, {"keywords": ["legal", "contract", "terms", "privacy", "policy"], "task": "Review legal terms and privacy policy updates", "priority": "medium", "owner": "Legal", "deadline": "T-15 days"}, ] tasks: list[dict] = [] for tmpl in task_templates: if any(kw in text for kw in tmpl["keywords"]): tasks.append({ "task": tmpl["task"], "priority": tmpl["priority"], "owner": tmpl["owner"], "deadline": tmpl["deadline"], }) # Always include goal definition as the first task always = { "task": "Define launch goals and success metrics", "priority": "high", "owner": "Product", "deadline": "T-30 days", } if not any(t["task"] == always["task"] for t in tasks): tasks.insert(0, always) # Always include a launch-day verification task launch_day = { "task": "Launch-day verification and monitoring", "priority": "high", "owner": "Engineering", "deadline": "Launch day (T-0)", } if not any(t["task"] == launch_day["task"] for t in tasks): tasks.append(launch_day) return tasks # --------------------------------------------------------------------------- # 2. check_readiness # --------------------------------------------------------------------------- @function_tool def check_readiness(tasks: str, launch_date: str) -> dict: """Check launch readiness against a five-point rubric. The rubric covers: code frozen, docs written, QA passed, marketing ready, and support staffed. We look for evidence of each in the provided task list text. Args: tasks: The extracted task list (as a string or JSON). launch_date: Target launch date in YYYY-MM-DD format. Returns: {ready: bool, score: int (0-100), gaps: list[str]} """ text = (tasks or "").lower() rubric = [ ("code frozen", ["code", "freeze", "tag", "release candidate", "engineering"]), ("docs written", ["doc", "documentation", "guide", "readme", "writing"]), ("QA passed", ["qa", "test", "regression", "smoke", "bug"]), ("marketing ready", ["market", "campaign", "social", "ad", "press", "blog"]), ("support staffed", ["support", "customer", "runbook", "oncall", "help"]), ] covered: list[str] = [] gaps: list[str] = [] for label, keywords in rubric: if any(kw in text for kw in keywords): covered.append(label) else: gaps.append(label) score = int((len(covered) / len(rubric)) * 100) ready = len(gaps) == 0 and score >= 100 # If launch date is missing we flag it as a gap too if not launch_date or not launch_date.strip(): gaps.append("launch date not set") score = max(0, score - 10) return { "ready": ready, "score": score, "gaps": gaps, "covered": covered, } # --------------------------------------------------------------------------- # 3. generate_checklist # --------------------------------------------------------------------------- @function_tool def generate_checklist(tasks: str) -> list[dict]: """Generate per-owner checklists from a task list. Parses the tasks string and groups items by owner, producing a checklist for each owner with concrete action items. Args: tasks: The extracted task list (string or JSON). Returns: A list of {owner, items: [str]} dicts. """ # Try to parse as JSON list of dicts; fall back to line parsing import json task_items: list[dict] = [] stripped = (tasks or "").strip() if stripped: try: parsed = json.loads(stripped) if isinstance(parsed, list): task_items = parsed except (json.JSONDecodeError, TypeError): # Fall back: split by newline, treat each line as a task for line in stripped.splitlines(): line = line.strip().strip("-").strip() if line: task_items.append({"task": line, "owner": "General"}) # If nothing parsed, provide a default checklist if not task_items: task_items = [ {"task": "Define launch goals and metrics", "owner": "Product"}, {"task": "Freeze code and tag release candidate", "owner": "Engineering"}, {"task": "Complete QA regression testing", "owner": "QA"}, {"task": "Write documentation", "owner": "Technical Writing"}, {"task": "Prepare marketing campaign", "owner": "Marketing"}, {"task": "Staff support team", "owner": "Customer Success"}, ] # Group tasks by owner by_owner: dict[str, list[str]] = {} for t in task_items: owner = t.get("owner", "General") task_name = t.get("task", str(t)) by_owner.setdefault(owner, []).append(task_name) # Build owner checklists with concrete action items checklists: list[dict] = [] for owner, task_names in by_owner.items(): items: list[str] = [] for tn in task_names: items.append(f"Confirm: {tn}") items.append(f"Sign off: {tn}") # Add a shared pre-launch check for every owner items.append("Review final launch checklist with launch lead") checklists.append({"owner": owner, "items": items}) # Ensure there is always a Launch Lead checklist if not any(c["owner"] == "Launch Lead" for c in checklists): checklists.append({ "owner": "Launch Lead", "items": [ "Confirm all owners have signed off", "Verify go/no-go criteria are met", "Schedule launch-day standup", "Monitor dashboards during launch window", ], }) return checklists # --------------------------------------------------------------------------- # 4. draft_copy # --------------------------------------------------------------------------- @function_tool def draft_copy(channel: str, product: str, audience: str) -> dict: """Draft marketing copy for a specific channel. Generates a headline, body, and call-to-action tailored to the channel, product, and audience. Args: channel: Marketing channel (e.g. "email", "twitter", "blog", "linkedin", "landing_page"). product: Product name or short description. audience: Target audience description. Returns: {headline: str, body: str, cta: str} """ ch = (channel or "").lower().strip() prod = (product or "").strip() or "our new product" aud = (audience or "").strip() or "your team" # Channel-specific copy templates if ch in ("twitter", "x", "tweet"): headline = f"🚀 {prod} is live — built for {aud}." body = ( f"We just shipped {prod}, designed from the ground up for {aud}. " "Faster, simpler, and ready for production today. " "Check it out 👇" ) cta = "Try it now →" elif ch in ("email", "newsletter", "drip"): headline = f"Introducing {prod} — now available for {aud}" body = ( f"Hi there,\n\n" f"We're excited to announce {prod}, purpose-built for {aud}. " "This release brings a refined experience, improved performance, " "and the features you've been asking for.\n\n" "Here's what's new:\n" " • Streamlined workflow that saves you time\n" " • Built-in reliability for production use\n" " • Thoughtful design choices throughout\n\n" "We'd love your feedback as you try it out." ) cta = "Get started today" elif ch in ("blog", "blog post", "announcement"): headline = f"{prod}: A new standard for {aud}" body = ( f"Today we're launching {prod}, our most refined release yet. " f"Built specifically for {aud}, it combines the features you need " "with the reliability you expect.\n\n" "In this post we'll walk through what's new, why we built it, " "and how to get started in under five minutes." ) cta = "Read the full announcement" elif ch in ("linkedin", "li"): headline = f"We just launched {prod} — here's why it matters for {aud}" body = ( f"After months of iteration, {prod} is now generally available. " f"We built it for {aud} who need a solution that's both powerful " "and dependable. Here's what went into it and what it means for you." ) cta = "Learn more" elif ch in ("landing", "landing page", "landing_page", "homepage"): headline = f"{prod} — purpose-built for {aud}" body = ( f"{prod} gives {aud} the tools they need to move faster, " "with the reliability required for real production workloads. " "No noise, no bloat — just what matters." ) cta = "Start building for free" else: # Generic fallback for unknown channels headline = f"Introducing {prod}" body = ( f"{prod} is now available. Designed for {aud}, it delivers " "the features and reliability you need to succeed." ) cta = "Learn more" return { "headline": headline, "body": body, "cta": cta, "channel": ch or "general", }