Spaces:
Paused
Paused
File size: 7,887 Bytes
b484b4d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 | """
Launch Desk — FastAPI Backend
==============================
Serves the frontend and exposes an SSE streaming endpoint that runs
the Launch Planner agent via the OpenAI Agents SDK.
"""
from __future__ import annotations
import json
import os
from pathlib import Path
from typing import Any
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse
from pydantic import BaseModel, Field
from agents import Runner
from agents.items import (
ToolCallItem,
ToolCallOutputItem,
)
from agents.stream_events import (
RawResponsesStreamEvent,
RunItemStreamEvent,
)
from server.agent import agent
# ---------------------------------------------------------------------------
# App setup
# ---------------------------------------------------------------------------
app = FastAPI(title="Launch Desk")
FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
# ---------------------------------------------------------------------------
# Request model
# ---------------------------------------------------------------------------
class LaunchBrief(BaseModel):
brief: str = Field(default="", description="Product brief text")
audience: str = Field(default="", description="Target audience")
launch_date: str = Field(default="", description="Target launch date YYYY-MM-DD")
constraints: str = Field(default="", description="Constraints or requirements")
assets: str = Field(default="", description="Available assets")
# ---------------------------------------------------------------------------
# SSE helper
# ---------------------------------------------------------------------------
def _sse(data: dict) -> str:
"""Format a dict as a Server-Sent Event line."""
return f"data: {json.dumps(data)}\n\n"
def _build_prompt(req: LaunchBrief) -> str:
"""Compose the user prompt from the structured request."""
parts: list[str] = []
if req.brief:
parts.append(f"## Product Brief\n{req.brief}")
if req.audience:
parts.append(f"## Target Audience\n{req.audience}")
if req.launch_date:
parts.append(f"## Launch Date\n{req.launch_date}")
if req.constraints:
parts.append(f"## Constraints\n{req.constraints}")
if req.assets:
parts.append(f"## Available Assets\n{req.assets}")
if not parts:
parts.append(
"No launch details were provided. Ask the user for a product "
"brief, audience, and target launch date."
)
parts.append(
"\n## Instructions\n"
"1. Use extract_tasks to break the brief into prioritised tasks.\n"
"2. Use check_readiness to assess launch readiness.\n"
"3. Use generate_checklist to create per-owner checklists.\n"
"4. Use draft_copy to write launch copy for email and twitter.\n"
"5. Produce a final plan with these sections:\n"
" - Prioritized Plan\n"
" - Risk Register\n"
" - Owner Checklist\n"
" - Launch Copy\n"
" - Follow-up Questions"
)
return "\n\n".join(parts)
# ---------------------------------------------------------------------------
# SSE streaming generator
# ---------------------------------------------------------------------------
async def _stream_agent(req: LaunchBrief):
"""Async generator yielding SSE events from the agent run."""
prompt = _build_prompt(req)
# Track which tool calls we've seen so we can pair start/end events.
# call_id -> tool_name
seen_calls: dict[str, str] = {}
try:
result = Runner.run_streamed(agent, input=prompt)
async for event in result.stream_events():
# --- Raw LLM streaming events (text deltas) ---
if isinstance(event, RawResponsesStreamEvent):
raw = event.data
# ResponseTextDeltaEvent has type "response.output_text.delta"
if hasattr(raw, "type") and raw.type == "response.output_text.delta":
delta_text = getattr(raw, "delta", "")
if delta_text:
yield _sse({"type": "text_delta", "text": delta_text})
continue
# --- Run item events (tool calls, tool outputs, messages) ---
if isinstance(event, RunItemStreamEvent):
item = event.item
# Tool call started
if isinstance(item, ToolCallItem):
tool_name = item.tool_name or "unknown_tool"
call_id = item.call_id or ""
if call_id:
seen_calls[call_id] = tool_name
# Try to extract arguments from the raw item
args: Any = {}
raw_item = item.raw_item
if isinstance(raw_item, dict):
args = raw_item.get("arguments", {})
elif hasattr(raw_item, "arguments"):
args = raw_item.arguments
# Parse args if they're a JSON string
if isinstance(args, str):
try:
args = json.loads(args)
except (json.JSONDecodeError, TypeError):
pass
yield _sse({
"type": "tool_start",
"name": tool_name,
"args": args,
})
continue
# Tool call completed
if isinstance(item, ToolCallOutputItem):
raw_item = item.raw_item
call_id = item.call_id or ""
tool_name = seen_calls.get(call_id, "unknown_tool")
output = item.output
# Try to parse output as JSON for nicer SSE payload
if isinstance(output, str):
try:
output = json.loads(output)
except (json.JSONDecodeError, TypeError):
pass
yield _sse({
"type": "tool_end",
"name": tool_name,
"result": output,
})
continue
# --- Done ---
final_output = result.final_output
yield _sse({"type": "done", "output": final_output})
except Exception as exc:
yield _sse({
"type": "error",
"message": f"Agent error: {exc}",
})
# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
@app.get("/api/health")
async def health():
return {"status": "ok"}
@app.post("/api/agent/stream")
async def agent_stream(req: LaunchBrief):
return StreamingResponse(
_stream_agent(req),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
},
)
@app.get("/", response_class=HTMLResponse)
async def serve_frontend():
index_path = FRONTEND_DIR / "index.html"
if index_path.exists():
return HTMLResponse(content=index_path.read_text(encoding="utf-8"))
return HTMLResponse(content="<h1>Frontend not found</h1>", status_code=404)
# ---------------------------------------------------------------------------
# Entry point for `python -m server.main` or `uvicorn server.main:app`
# ---------------------------------------------------------------------------
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"server.main:app",
host="0.0.0.0",
port=8001,
reload=False,
)
|