diff --git a/pyproject.toml b/pyproject.toml index f94eb859..47b171e3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,6 +14,7 @@ dependencies = [ "requests>=2.0, <3", "types-requests>=2.0, <3", "mcp>=1.6.0, <2; python_version >= '3.10'", + ] classifiers = [ "Typing :: Typed", diff --git a/src/agents/agent_onboarding.py b/src/agents/agent_onboarding.py new file mode 100644 index 00000000..18eb7aa3 --- /dev/null +++ b/src/agents/agent_onboarding.py @@ -0,0 +1,83 @@ +import sys +import os +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from fastapi import APIRouter, Request +from agents import Agent, Runner +from datetime import datetime +import json +import httpx + +router = APIRouter() + +# Define the onboarding agent +onboarding_agent = Agent( + name="OnboardingAgent", + instructions=""" +You are an onboarding assistant helping new influencers introduce themselves. +Your job is to: +1. Gently ask about their interests, content goals, and style. +2. Summarize their early profile with a few soft fields. +3. Optionally ask a clarifying follow-up question if needed. + +Return your response in the following format: +{ + "output_type": "soft_profile", + "contains_image": false, + "details": { + "interests": ["wellness", "fitness"], + "preferred_style": "authentic, relaxed", + "content_goals": "collaborations, brand storytelling", + "next_question": "Would you consider doing live sessions?" + } +} +Only reply in this format. +""" +) + +@router.post("/onboard") +async def onboard_influencer(request: Request): + data = await request.json() + user_input = data.get("input", "") + user_id = data.get("user_id", "anonymous") + webhook_url = data.get("webhook_url") + debug_info = {} + + result = await Runner.run(onboarding_agent, input=user_input) + + try: + parsed_output = json.loads(result.final_output) + output_type = parsed_output.get("output_type") + output_details = parsed_output.get("details") + contains_image = parsed_output.get("contains_image", False) + + if not output_type or not output_details: + raise ValueError("Missing required output keys") + except Exception as e: + parsed_output = None + output_type = "raw_text" + output_details = result.final_output + contains_image = False + debug_info["validation_error"] = str(e) + debug_info["raw_output"] = result.final_output + + session = { + "agent_type": "onboarding", + "user_id": user_id, + "output_type": output_type, + "contains_image": contains_image, + "output_details": output_details, + "created_at": datetime.utcnow().isoformat(), + } + + if debug_info: + session["debug_info"] = debug_info + + if webhook_url: + async with httpx.AsyncClient() as client: + try: + await client.post(webhook_url, json=session) + except Exception as e: + session["webhook_error"] = str(e) + + return session diff --git a/src/agents/agent_profilebuilder.py b/src/agents/agent_profilebuilder.py new file mode 100644 index 00000000..8bb6b28c --- /dev/null +++ b/src/agents/agent_profilebuilder.py @@ -0,0 +1,100 @@ +import sys +import os +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from fastapi import APIRouter, Request +from agents import Agent, Runner +from agents.tool import WebSearchTool +from datetime import datetime +import json +import httpx + +router = APIRouter() + +# Predefined Bubble webhook URL +WEBHOOK_URL = "https://helpmeaiai.bubbleapps.io/version-test/api/1.1/wf/openai_profilebuilder_return" + +# ProfileBuilder agent skeleton; tools will be set per-request for dynamic locale/fallback +profile_builder_agent = Agent( + name="ProfileBuilderAgent", + instructions=""" +You are a profile builder assistant with web search capability. + +You will receive a set of key-value inputs (e.g., profile_uuid, handle URL, etc.). +Your job: +1. Use the provided fields (including fallback follower count if given). +2. If a locale is provided, use it to tailor the web search tool's user_location. +3. Perform web searches and reasoning to determine follower_count, posting_style, industry, engagement_rate, and any notable public context. +4. Summarize this into JSON as follows: +{ + "output_type": "structured_profile", + "contains_image": false, + "details": { + "profile_uuid": "...", + "summary": "Concise profile summary...", + "prompt_snippet": { "tone": "...", "goal": "...", "platform": "..." }, + "follower_count": 12345, + "posting_style": "...", + "industry": "...", + "engagement_rate": "...", + "additional_context": "..." + } +} +Only return JSON with exactly these fields—no markdown or commentary. +""", + tools=[] +) + +@router.post("/profilebuilder") +async def build_profile(request: Request): + data = await request.json() + # Extract core identifiers and optional fallbacks + profile_uuid = data.pop("profile_uuid", None) + provided_fc = data.pop("provided_follower_count", None) + locale_text = data.pop("locale", None) + + # Build tool list dynamically based on locale + user_loc = {"type": "approximate", "region": locale_text} if locale_text else None + tools = [WebSearchTool(user_location=user_loc, search_context_size="low")] + profile_builder_agent.tools = tools + + # Flatten remaining inputs into prompt lines + prompt_lines = [] + for key, val in data.items(): + if val not in (None, "", "null"): + prompt_lines.append(f"{key}: {val}") + if provided_fc is not None: + prompt_lines.append(f"Provided follower count: {provided_fc}") + + # Construct the agent prompt + agent_input = f"Profile UUID: {profile_uuid}\n" + "\n".join(prompt_lines) + + # Invoke the agent + result = await Runner.run(profile_builder_agent, input=agent_input) + + # Clean markdown fences + output = result.final_output.strip() + if output.startswith("```") and output.endswith("```"): + output = output.split("\n", 1)[-1].rsplit("\n", 1)[0] + + # Parse agent JSON response + try: + parsed = json.loads(output) + details = parsed.get("details", {}) + except Exception: + details = {} + + # Build profile_data payload dynamically + profile_data = {"profile_uuid": profile_uuid} + for k, v in details.items(): + profile_data[k] = v + profile_data["created_at"] = datetime.utcnow().isoformat() + + # Post to Bubble webhook + async with httpx.AsyncClient() as client: + try: + await client.post(WEBHOOK_URL, json=profile_data) + except Exception as e: + profile_data["webhook_error"] = str(e) + + return profile_data diff --git a/src/agents/agent_server.py b/src/agents/agent_server.py new file mode 100644 index 00000000..327279ab --- /dev/null +++ b/src/agents/agent_server.py @@ -0,0 +1,185 @@ +# File: src/agents/agent_server.py + +import sys +import os +from dotenv import load_dotenv + +# 1) Load environment variables from .env +load_dotenv() + +# 2) Ensure src/ is on the Python path so “util” is importable +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from fastapi import FastAPI, Request, HTTPException +from fastapi.middleware.cors import CORSMiddleware +from pydantic import parse_obj_as, ValidationError +import json +from datetime import datetime +import httpx + +# 3) Core SDK imports +from agents import Agent, Runner, tool + +# 4) Pydantic schemas and service handlers +from util.schemas import Inbound # Union of NewTask, NewMessage +from util.services import handle_new_task, handle_new_message + +# ─────────────────────────────────────────────────────────── +# 5) Agent definitions (Phase 1: keep here for simplicity) +# ─────────────────────────────────────────────────────────── +# Manager: routes requests or asks for clarifications +manager_agent = Agent( + name="Manager", + instructions=""" +You are an intelligent router for user requests. +Decide the intent behind the message: strategy, content, repurpose, feedback. +If you are unsure or need more info, ask a clarifying question instead of routing. +Respond in strict JSON like: +{ "route_to": "strategy", "reason": "User wants a campaign plan" } +""" +) + +# Strategy: builds a 7‑day social campaign plan +strategy_agent = Agent( + name="StrategyAgent", + instructions=""" +You create clear, actionable 7-day social media campaign strategies. +If user input is unclear or missing platform, audience, or tone — ask for clarification. +Respond in structured JSON like: +{ + "output_type": "strategy_plan", + "contains_image": false, + "details": { + "days": [ + { "title": "...", "theme": "...", "cta": "..." } + ] + } +} +Only return JSON in this format. +""", + tools=[] +) + +# Content: writes social post variants +content_agent = Agent( + name="ContentAgent", + instructions=""" +You write engaging, brand-aligned social content. +If user input lacks platform or goal, ask for clarification. +Return post drafts in this JSON format: +{ + "output_type": "content_variants", + "contains_image": false, + "details": { + "variants": [ + { + "platform": "Instagram", + "caption": "...", + "hook": "...", + "cta": "..." + } + ] + } +} +Only respond in this format. +""", + tools=[] +) + +# Repurpose: converts posts into new formats +repurpose_agent = Agent( + name="RepurposeAgent", + instructions=""" +You convert existing posts into new formats for different platforms. +Respond using this format: +{ + "output_type": "repurposed_posts", + "contains_image": false, + "details": { + "original": "...", + "repurposed": [ + { + "platform": "...", + "caption": "...", + "format": "..." + } + ] + } +} +""", + tools=[] +) + +# Feedback: critiques content and suggests improvements +feedback_agent = Agent( + name="FeedbackAgent", + instructions=""" +You evaluate content and offer improvements. +Respond in this structured format: +{ + "output_type": "content_feedback", + "contains_image": false, + "details": { + "original": "...", + "feedback": "...", + "suggested_edit": "..." + } +} +""", + tools=[] +) + +# Map Manager’s routing keys to Agent instances +AGENT_MAP = { + "strategy": strategy_agent, + "content": content_agent, + "repurpose": repurpose_agent, + "feedback": feedback_agent, +} +# ─────────────────────────────────────────────────────────── + +# 6) Instantiate FastAPI +app = FastAPI() + +# 7) CORS middleware (adjust allow_origins as needed) +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +# 8) Include your existing agent routers +from .agent_onboarding import router as onboarding_router +from .agent_profilebuilder import router as profilebuilder_router + +app.include_router(onboarding_router) +app.include_router(profilebuilder_router) + +# 9) Unified /agent endpoint +@app.post("/agent") +async def agent_endpoint(request: Request): + """ + Handles all client calls: + - action = "new_task" + - action = "new_message" + - future actions as you add them to Inbound + """ + body = await request.json() + try: + payload = parse_obj_as(Inbound, body) + except ValidationError as e: + raise HTTPException(status_code=400, detail=e.errors()) + + if payload.action == "new_task": + return await handle_new_task(payload) + + elif payload.action == "new_message": + return await handle_new_message(payload) + + else: + raise HTTPException( + status_code=400, + detail=f"Unsupported action: {payload.action}" + ) diff --git a/src/agents/util/schemas.py b/src/agents/util/schemas.py new file mode 100644 index 00000000..c26c7ecb --- /dev/null +++ b/src/agents/util/schemas.py @@ -0,0 +1,17 @@ +from typing import Literal, Optional, Dict, Union +from pydantic import BaseModel, Field + +class NewTask(BaseModel): + action: Literal["new_task"] + task_type: str + user_prompt: str + params: Dict = Field(default_factory=dict) + first_agent: Optional[str] = "auto" + +class NewMessage(BaseModel): + action: Literal["new_message"] + task_id: str + message: str + agent_session_id: Optional[str] = None + +Inbound = Union[NewTask, NewMessage] diff --git a/src/agents/util/services.py b/src/agents/util/services.py new file mode 100644 index 00000000..d58b8dc0 --- /dev/null +++ b/src/agents/util/services.py @@ -0,0 +1,34 @@ +from your_orm_models import Task, AgentSession, Message # adapt import path +from util.webhook import post_webhook, STRUCTURED_URL, CLARIFICATION_URL +from agents.runner import run_agent, decide_session # your existing runner + +async def handle_new_task(p: NewTask): + task = Task.create( + user_id=p.request_user.id, + title=p.user_prompt[:40], + type=p.task_type, + status="pending", + params=p.params, + ) + first_def = "manager" if p.first_agent == "auto" else p.first_agent + session = AgentSession.create( + task=task, agent_definition=first_def, status="running" + ) + Message.create(task=task, role="user", content=p.user_prompt) + await run_agent(session) + return {"task_id": task.id} + +async def handle_new_message(p: NewMessage): + Message.create( + task_id=p.task_id, + agent_session_id=p.agent_session_id, + role="user", + content=p.message, + ) + session = ( + AgentSession.get(p.agent_session_id) + if p.agent_session_id + else decide_session(p.task_id) + ) + await run_agent(session) + return {"ok": True} diff --git a/src/agents/util/webhook.py b/src/agents/util/webhook.py new file mode 100644 index 00000000..825800d5 --- /dev/null +++ b/src/agents/util/webhook.py @@ -0,0 +1,16 @@ +import os, asyncio, logging, httpx + +STRUCTURED_URL = os.getenv("BUBBLE_STRUCTURED_URL") +CLARIFICATION_URL = os.getenv("BUBBLE_CHAT_URL") + +async def post_webhook(url: str, data: dict, retries: int = 3): + for i in range(retries): + try: + async with httpx.AsyncClient(timeout=10) as client: + r = await client.post(url, json=data) + r.raise_for_status() + return + except Exception as e: + if i == retries - 1: + logging.error("Webhook failed %s: %s", url, e) + await asyncio.sleep(2 ** i)