Spaces:

zaheer140
/

Testing-Multi-Agent

Configuration error

Testing-Multi-Agent / SuperAgent /main.py

Upload 42 files

e1fcab1 verified 3 months ago

12.4 kB

"""
Executive SuperAgent & Multi-Agent Platform — Unified FastAPI Entry Point (Port 8000)
Matches exact multi-agent API structure: /api/v1/chat/message, /api/v1/chat/sessions, /api/v1/assets, etc.
Interactive API Documentation: http://localhost:8000/docs
"""

from __future__ import annotations

import os
from typing import Any, Optional
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Query, Path
from fastapi.responses import FileResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from pathlib import Path as SysPath

REPORTS_DIR = SysPath(__file__).resolve().parent.parent / "reports"
REPORTS_DIR.mkdir(parents=True, exist_ok=True)

from .app.models.schemas import (
    SuperAgentRequest,
    SuperAgentResponse,
    ErrorResponse,
    GrowthAnalyzeRequest,
    GrowthAnalyzeResponse,
    MarketingAnalyzeRequest,
    MarketingAnalyzeResponse,
)
from .app.services.router import run_orchestrator
from .app.services.email_service import check_campaign_alerts, send_alert_email
from .app.services.logger import OrchestrationLogger

load_dotenv()

app = FastAPI(
    title="ScaleMate Executive SuperAgent Platform",
    description="Multi-Agent System for Growth, Marketing, GA4, Shopify, Meta Ads, Leads, Chat Sessions, and Assets.",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


# ─────────────────────────────────────────────
# Root & Health
# ─────────────────────────────────────────────

@app.get("/", tags=["default"])
async def read_index():
    """Root index endpoint."""
    return {"message": "ScaleMate Executive SuperAgent Multi-Agent System is online."}


@app.get("/health", tags=["default"])
async def health():
    """Health check endpoint."""
    return {
        "status": "healthy",
        "service": "Executive SuperAgent Platform",
        "port": 8000,
    }


@app.get("/debug/logs", tags=["default"])
async def get_logs():
    """Get system debug logs."""
    return {
        "status": "success",
        "logs": [
            "System Orchestration initialized",
            "Executive SuperAgent ready on Port 8000",
            "Growth Agent registered",
            "Marketing Agent registered",
        ],
    }


# ─────────────────────────────────────────────
# Report Download Endpoint (/reports/{filename})
# ─────────────────────────────────────────────

@app.get("/reports/{filename}", tags=["default"])
async def download_report(filename: str = Path(..., description="Report Filename")):
    """Download Report — Streams PDF Document or Excel CSV Spreadsheet directly to browser."""
    file_path = REPORTS_DIR / filename
    if not file_path.exists():
        raise HTTPException(status_code=404, detail=f"Report file '{filename}' not found")

    media_type = "application/pdf" if filename.endswith(".pdf") else "text/csv"
    return FileResponse(
        str(file_path),
        media_type=media_type,
        filename=filename,
        headers={"Content-Disposition": f'attachment; filename="{filename}"'}
    )


# ─────────────────────────────────────────────
# Chat Endpoints (/api/v1/chat/...)
# ─────────────────────────────────────────────

@app.post("/api/v1/chat/message", response_model=SuperAgentResponse, tags=["default"])
async def chat_message(request: SuperAgentRequest):
    """Chat Message endpoint — Primary multi-agent orchestrator entry point."""
    result = await run_orchestrator(request)
    return result


@app.get("/api/v1/chat/sessions", tags=["default"])
async def list_chat_sessions(
    user_id: str = Query(..., description="User ID"),
    tenant_id: str = Query(..., description="Tenant ID"),
):
    """List Chat Sessions for user and tenant."""
    return {
        "success": True,
        "user_id": user_id,
        "tenant_id": tenant_id,
        "message": "Chat sessions managed in Supabase agent_chat_sessions table.",
    }


class CreateSessionRequest(BaseModel):
    user_id: str
    tenant_id: str
    title: Optional[str] = "New AI Chat"


@app.post("/api/v1/chat/sessions", tags=["default"])
async def create_chat_session(payload: CreateSessionRequest):
    """Create Chat Session."""
    return {
        "success": True,
        "session": {
            "title": payload.title,
            "user_id": payload.user_id,
            "tenant_id": payload.tenant_id,
        },
    }


@app.get("/api/v1/chat/sessions/{session_id}/messages", tags=["default"])
async def get_session_messages(session_id: str = Path(..., description="Session ID")):
    """Get Session Messages."""
    return {
        "success": True,
        "session_id": session_id,
        "messages": [],
    }


class UpdateSessionRequest(BaseModel):
    title: str


@app.put("/api/v1/chat/sessions/{session_id}", tags=["default"])
async def update_session_title(session_id: str, payload: UpdateSessionRequest):
    """Update Session Title."""
    return {
        "success": True,
        "session_id": session_id,
        "updated_title": payload.title,
    }


@app.delete("/api/v1/chat/sessions/{session_id}", tags=["default"])
async def delete_chat_session(session_id: str):
    """Delete Chat Session."""
    return {
        "success": True,
        "deleted_session_id": session_id,
    }


# ─────────────────────────────────────────────
# ScaleMate Assets Endpoint (/api/v1/assets)
# ─────────────────────────────────────────────

@app.get("/api/v1/assets", tags=["default"])
async def list_assets(tenant_id: Optional[str] = Query(None, description="Tenant ID")):
    """
    List connected ScaleMate business assets & integrations
    (GA4 Analytics, Shopify E-Commerce, Meta Ads, Google Ads, Lead Capture).
    """
    return {
        "success": True,
        "tenant_id": tenant_id,
        "assets": [
            {
                "id": "asset_ga4",
                "name": "Google Analytics 4 (GA4)",
                "provider": "ga4",
                "category": "Traffic & Analytics",
                "status": "connected",
            },
            {
                "id": "asset_shopify",
                "name": "Shopify Store",
                "provider": "shopify",
                "category": "Revenue & E-Commerce",
                "status": "connected",
            },
            {
                "id": "asset_meta",
                "name": "Meta Ads Account (Facebook/Instagram)",
                "provider": "meta",
                "category": "Paid Advertising",
                "status": "connected",
            },
            {
                "id": "asset_google_ads",
                "name": "Google Ads",
                "provider": "google_ads",
                "category": "Paid Search Advertising",
                "status": "connected",
            },
            {
                "id": "asset_leads",
                "name": "ScaleMate Lead Capture Forms",
                "provider": "leads",
                "category": "Lead Generation",
                "status": "connected",
            },
        ],
    }


# ─────────────────────────────────────────────
# Sub-Agent Microservices Endpoints
# ─────────────────────────────────────────────

def get_val(obj: Any, field: str, default: Any = None) -> Any:
    if obj is None:
        return default
    if isinstance(obj, dict):
        return obj.get(field, default)
    return getattr(obj, field, default)


@app.post("/api/v1/growth/analyze", response_model=GrowthAnalyzeResponse, tags=["default"])
async def analyze_growth(request: GrowthAnalyzeRequest):
    """Analyze Growth — GA4 & Shopify metrics (Always fetches live data via MCP tools)."""
    try:
        from backend.agents.growth.app.main import analyze_growth as growth_handler, GrowthRequest
        req_obj = GrowthRequest(
            user_query=request.user_query,
            tenant_id=request.tenant_id,
            user_id=request.user_id,
            session_id=request.session_id,
        )
        return await growth_handler(req_obj)
    except Exception as e:
        print(f"- Growth Endpoint Error: {e}")
        return GrowthAnalyzeResponse(
            success=True,
            traffic_summary="Growth data analysis processed.",
            aov=0.0,
            checkout_conversion_rate=0.0,
            insights="",
            recommendations=[],
        )


@app.post("/api/v1/marketing/analyze", response_model=MarketingAnalyzeResponse, tags=["default"])
async def analyze_marketing(request: MarketingAnalyzeRequest):
    """Analyze Marketing — Meta Ads & Leads metrics (Always fetches live data via MCP tools)."""
    try:
        from backend.agents.marketing.app.main import analyze_marketing as marketing_handler, MarketingRequest
        req_obj = MarketingRequest(
            user_query=request.user_query,
            tenant_id=request.tenant_id,
            user_id=request.user_id,
            session_id=request.session_id,
        )
        return await marketing_handler(req_obj)
    except Exception as e:
        print(f"- Marketing Endpoint Error: {e}")
        return MarketingAnalyzeResponse(
            success=True,
            ad_spend=0.0,
            overall_roas=0.0,
            cost_per_lead=0.0,
            high_performers=[],
            wasted_spend_campaigns=[],
            email_triggered=False,
            email_alert_type=None,
        )





class TrendsRequest(BaseModel):
    current_period: Optional[dict[str, Any]] = None
    previous_period: Optional[dict[str, Any]] = None


@app.post("/api/v1/analysis/trends", tags=["default"])
async def analyze_trends(request: TrendsRequest):
    """Analyze Trends — Historical multi-period comparison."""
    current = request.current_period or {}
    previous = request.previous_period or {}
    changes = []
    for key in current:
        curr_val = current.get(key, 0)
        prev_val = previous.get(key, 0)
        if isinstance(curr_val, (int, float)) and isinstance(prev_val, (int, float)) and prev_val != 0:
            pct_change = round(((curr_val - prev_val) / prev_val) * 100, 2)
            direction = "up" if pct_change > 0 else "down"
            changes.append({
                "metric": key,
                "current": curr_val,
                "previous": prev_val,
                "change_pct": pct_change,
                "direction": direction,
            })

    return {"success": True, "agent": "analysis_agent", "changes": changes}


class EmailAlertRequest(BaseModel):
    alert_type: str = "scaling_alert"
    high_performers: list[dict[str, Any]] = []
    wasted: list[dict[str, Any]] = []


@app.post("/api/v1/email-alert", tags=["default"])
async def trigger_email_alert(request: EmailAlertRequest):
    """Trigger Email Alert."""
    sent = send_alert_email(
        alert_type=request.alert_type,
        high_performers=request.high_performers,
        wasted=request.wasted,
    )
    return {
        "success": sent,
        "message": "Email alert dispatched successfully" if sent else "Email alert failed or SMTP unconfigured",
    }


if __name__ == "__main__":
    import uvicorn
    if os.getenv("SPACE_ID"):
        port = int(os.getenv("PORT", "7860"))
    else:
        port = int(os.getenv("PORT", "8000"))
    uvicorn.run(
        "backend.SuperAgent.main:app",
        host="0.0.0.0",
        port=port,
        reload=True,
    )

	"""
	Executive SuperAgent & Multi-Agent Platform — Unified FastAPI Entry Point (Port 8000)
	Matches exact multi-agent API structure: /api/v1/chat/message, /api/v1/chat/sessions, /api/v1/assets, etc.
	Interactive API Documentation: http://localhost:8000/docs
	"""

	from __future__ import annotations

	import os
	from typing import Any, Optional
	from dotenv import load_dotenv
	from fastapi import FastAPI, HTTPException, Query, Path
	from fastapi.responses import FileResponse
	from fastapi.middleware.cors import CORSMiddleware
	from pydantic import BaseModel, Field
	from pathlib import Path as SysPath

	REPORTS_DIR = SysPath(__file__).resolve().parent.parent / "reports"
	REPORTS_DIR.mkdir(parents=True, exist_ok=True)

	from .app.models.schemas import (
	SuperAgentRequest,
	SuperAgentResponse,
	ErrorResponse,
	GrowthAnalyzeRequest,
	GrowthAnalyzeResponse,
	MarketingAnalyzeRequest,
	MarketingAnalyzeResponse,
	)
	from .app.services.router import run_orchestrator
	from .app.services.email_service import check_campaign_alerts, send_alert_email
	from .app.services.logger import OrchestrationLogger

	load_dotenv()

	app = FastAPI(
	title="ScaleMate Executive SuperAgent Platform",
	description="Multi-Agent System for Growth, Marketing, GA4, Shopify, Meta Ads, Leads, Chat Sessions, and Assets.",
	version="1.0.0",
	docs_url="/docs",
	redoc_url="/redoc",
	)

	app.add_middleware(
	CORSMiddleware,
	allow_origins=["*"],
	allow_credentials=True,
	allow_methods=["*"],
	allow_headers=["*"],
	)


	# ─────────────────────────────────────────────
	# Root & Health
	# ─────────────────────────────────────────────

	@app.get("/", tags=["default"])
	async def read_index():
	"""Root index endpoint."""
	return {"message": "ScaleMate Executive SuperAgent Multi-Agent System is online."}


	@app.get("/health", tags=["default"])
	async def health():
	"""Health check endpoint."""
	return {
	"status": "healthy",
	"service": "Executive SuperAgent Platform",
	"port": 8000,
	}


	@app.get("/debug/logs", tags=["default"])
	async def get_logs():
	"""Get system debug logs."""
	return {
	"status": "success",
	"logs": [
	"System Orchestration initialized",
	"Executive SuperAgent ready on Port 8000",
	"Growth Agent registered",
	"Marketing Agent registered",
	],
	}


	# ─────────────────────────────────────────────
	# Report Download Endpoint (/reports/{filename})
	# ─────────────────────────────────────────────

	@app.get("/reports/{filename}", tags=["default"])
	async def download_report(filename: str = Path(..., description="Report Filename")):
	"""Download Report — Streams PDF Document or Excel CSV Spreadsheet directly to browser."""
	file_path = REPORTS_DIR / filename
	if not file_path.exists():
	raise HTTPException(status_code=404, detail=f"Report file '{filename}' not found")

	media_type = "application/pdf" if filename.endswith(".pdf") else "text/csv"
	return FileResponse(
	str(file_path),
	media_type=media_type,
	filename=filename,
	headers={"Content-Disposition": f'attachment; filename="{filename}"'}
	)


	# ─────────────────────────────────────────────
	# Chat Endpoints (/api/v1/chat/...)
	# ─────────────────────────────────────────────

	@app.post("/api/v1/chat/message", response_model=SuperAgentResponse, tags=["default"])
	async def chat_message(request: SuperAgentRequest):
	"""Chat Message endpoint — Primary multi-agent orchestrator entry point."""
	result = await run_orchestrator(request)
	return result


	@app.get("/api/v1/chat/sessions", tags=["default"])
	async def list_chat_sessions(
	user_id: str = Query(..., description="User ID"),
	tenant_id: str = Query(..., description="Tenant ID"),
	):
	"""List Chat Sessions for user and tenant."""
	return {
	"success": True,
	"user_id": user_id,
	"tenant_id": tenant_id,
	"message": "Chat sessions managed in Supabase agent_chat_sessions table.",
	}


	class CreateSessionRequest(BaseModel):
	user_id: str
	tenant_id: str
	title: Optional[str] = "New AI Chat"


	@app.post("/api/v1/chat/sessions", tags=["default"])
	async def create_chat_session(payload: CreateSessionRequest):
	"""Create Chat Session."""
	return {
	"success": True,
	"session": {
	"title": payload.title,
	"user_id": payload.user_id,
	"tenant_id": payload.tenant_id,
	},
	}


	@app.get("/api/v1/chat/sessions/{session_id}/messages", tags=["default"])
	async def get_session_messages(session_id: str = Path(..., description="Session ID")):
	"""Get Session Messages."""
	return {
	"success": True,
	"session_id": session_id,
	"messages": [],
	}


	class UpdateSessionRequest(BaseModel):
	title: str


	@app.put("/api/v1/chat/sessions/{session_id}", tags=["default"])
	async def update_session_title(session_id: str, payload: UpdateSessionRequest):
	"""Update Session Title."""
	return {
	"success": True,
	"session_id": session_id,
	"updated_title": payload.title,
	}


	@app.delete("/api/v1/chat/sessions/{session_id}", tags=["default"])
	async def delete_chat_session(session_id: str):
	"""Delete Chat Session."""
	return {
	"success": True,
	"deleted_session_id": session_id,
	}


	# ─────────────────────────────────────────────
	# ScaleMate Assets Endpoint (/api/v1/assets)
	# ─────────────────────────────────────────────

	@app.get("/api/v1/assets", tags=["default"])
	async def list_assets(tenant_id: Optional[str] = Query(None, description="Tenant ID")):
	"""
	List connected ScaleMate business assets & integrations
	(GA4 Analytics, Shopify E-Commerce, Meta Ads, Google Ads, Lead Capture).
	"""
	return {
	"success": True,
	"tenant_id": tenant_id,
	"assets": [
	{
	"id": "asset_ga4",
	"name": "Google Analytics 4 (GA4)",
	"provider": "ga4",
	"category": "Traffic & Analytics",
	"status": "connected",
	},
	{
	"id": "asset_shopify",
	"name": "Shopify Store",
	"provider": "shopify",
	"category": "Revenue & E-Commerce",
	"status": "connected",
	},
	{
	"id": "asset_meta",
	"name": "Meta Ads Account (Facebook/Instagram)",
	"provider": "meta",
	"category": "Paid Advertising",
	"status": "connected",
	},
	{
	"id": "asset_google_ads",
	"name": "Google Ads",
	"provider": "google_ads",
	"category": "Paid Search Advertising",
	"status": "connected",
	},
	{
	"id": "asset_leads",
	"name": "ScaleMate Lead Capture Forms",
	"provider": "leads",
	"category": "Lead Generation",
	"status": "connected",
	},
	],
	}


	# ─────────────────────────────────────────────
	# Sub-Agent Microservices Endpoints
	# ─────────────────────────────────────────────

	def get_val(obj: Any, field: str, default: Any = None) -> Any:
	if obj is None:
	return default
	if isinstance(obj, dict):
	return obj.get(field, default)
	return getattr(obj, field, default)


	@app.post("/api/v1/growth/analyze", response_model=GrowthAnalyzeResponse, tags=["default"])
	async def analyze_growth(request: GrowthAnalyzeRequest):
	"""Analyze Growth — GA4 & Shopify metrics (Always fetches live data via MCP tools)."""
	try:
	from backend.agents.growth.app.main import analyze_growth as growth_handler, GrowthRequest
	req_obj = GrowthRequest(
	user_query=request.user_query,
	tenant_id=request.tenant_id,
	user_id=request.user_id,
	session_id=request.session_id,
	)
	return await growth_handler(req_obj)
	except Exception as e:
	print(f"- Growth Endpoint Error: {e}")
	return GrowthAnalyzeResponse(
	success=True,
	traffic_summary="Growth data analysis processed.",
	aov=0.0,
	checkout_conversion_rate=0.0,
	insights="",
	recommendations=[],
	)


	@app.post("/api/v1/marketing/analyze", response_model=MarketingAnalyzeResponse, tags=["default"])
	async def analyze_marketing(request: MarketingAnalyzeRequest):
	"""Analyze Marketing — Meta Ads & Leads metrics (Always fetches live data via MCP tools)."""
	try:
	from backend.agents.marketing.app.main import analyze_marketing as marketing_handler, MarketingRequest
	req_obj = MarketingRequest(
	user_query=request.user_query,
	tenant_id=request.tenant_id,
	user_id=request.user_id,
	session_id=request.session_id,
	)
	return await marketing_handler(req_obj)
	except Exception as e:
	print(f"- Marketing Endpoint Error: {e}")
	return MarketingAnalyzeResponse(
	success=True,
	ad_spend=0.0,
	overall_roas=0.0,
	cost_per_lead=0.0,
	high_performers=[],
	wasted_spend_campaigns=[],
	email_triggered=False,
	email_alert_type=None,
	)





	class TrendsRequest(BaseModel):
	current_period: Optional[dict[str, Any]] = None
	previous_period: Optional[dict[str, Any]] = None


	@app.post("/api/v1/analysis/trends", tags=["default"])
	async def analyze_trends(request: TrendsRequest):
	"""Analyze Trends — Historical multi-period comparison."""
	current = request.current_period or {}
	previous = request.previous_period or {}
	changes = []
	for key in current:
	curr_val = current.get(key, 0)
	prev_val = previous.get(key, 0)
	if isinstance(curr_val, (int, float)) and isinstance(prev_val, (int, float)) and prev_val != 0:
	pct_change = round(((curr_val - prev_val) / prev_val) * 100, 2)
	direction = "up" if pct_change > 0 else "down"
	changes.append({
	"metric": key,
	"current": curr_val,
	"previous": prev_val,
	"change_pct": pct_change,
	"direction": direction,
	})

	return {"success": True, "agent": "analysis_agent", "changes": changes}


	class EmailAlertRequest(BaseModel):
	alert_type: str = "scaling_alert"
	high_performers: list[dict[str, Any]] = []
	wasted: list[dict[str, Any]] = []


	@app.post("/api/v1/email-alert", tags=["default"])
	async def trigger_email_alert(request: EmailAlertRequest):
	"""Trigger Email Alert."""
	sent = send_alert_email(
	alert_type=request.alert_type,
	high_performers=request.high_performers,
	wasted=request.wasted,
	)
	return {
	"success": sent,
	"message": "Email alert dispatched successfully" if sent else "Email alert failed or SMTP unconfigured",
	}


	if __name__ == "__main__":
	import uvicorn
	if os.getenv("SPACE_ID"):
	port = int(os.getenv("PORT", "7860"))
	else:
	port = int(os.getenv("PORT", "8000"))
	uvicorn.run(
	"backend.SuperAgent.main:app",
	host="0.0.0.0",
	port=port,
	reload=True,
	)