| from typing import Optional, List |
| from fastapi import APIRouter, Depends, HTTPException, Query |
| from sqlalchemy.orm import Session |
| from pydantic import BaseModel |
| from backend.database import get_db, Integration |
| from backend.auth import get_current_user, User |
| from backend.composio_client import get_composio |
|
|
| router = APIRouter() |
|
|
|
|
| class IntegrationResponse(BaseModel): |
| id: str |
| name: str |
| description: str |
| logo_url: str = "" |
| category: str = "Other" |
| auth_type: str = "api_key" |
| oauth_scopes: Optional[List[str]] = None |
| tool_count: int = 0 |
| trigger_count: int = 0 |
| is_active: bool = True |
|
|
|
|
| @router.get("", response_model=List[IntegrationResponse]) |
| async def list_integrations( |
| category: Optional[str] = None, |
| search: Optional[str] = None, |
| db: Session = Depends(get_db) |
| ): |
| composio = get_composio() |
| integrations = [] |
| if composio: |
| try: |
| raw = composio.toolkits.get() |
| for t in raw: |
| t_d = t |
| if hasattr(t, "model_dump"): |
| t_d = t.model_dump(mode="json") |
| elif hasattr(t, "__dict__"): |
| t_d = {k: v for k, v in t.__dict__.items() if not k.startswith("_")} |
| meta = t_d.get("meta", {}) or {} |
| cats = (meta.get("categories") or []) if isinstance(meta, dict) else [] |
| cat = cats[0].get("name", "Other") if cats else "Other" |
| if category and cat.lower() != category.lower(): |
| continue |
| name = t_d.get("name", t_d.get("slug", "")) |
| if search and search.lower() not in name.lower(): |
| continue |
| auth_schemes = t_d.get("auth_schemes") or [] |
| auth_type = "api_key" |
| if any("oauth" in str(s).lower() for s in auth_schemes): |
| auth_type = "oauth2" |
| integrations.append(IntegrationResponse( |
| id=t_d.get("slug", t_d.get("id", "")), |
| name=name, |
| description=t_d.get("description", ""), |
| logo_url="", |
| category=cat, |
| auth_type=auth_type, |
| tool_count=int(meta.get("tools_count", 0)) if isinstance(meta, dict) else 0, |
| )) |
| except Exception as e: |
| print(f"[Apps] SDK list failed: {e}") |
|
|
| if not integrations: |
| query = db.query(Integration).filter(Integration.is_active == True) |
| if category: |
| query = query.filter(Integration.category == category) |
| if search: |
| query = query.filter( |
| (Integration.name.ilike(f"%{search}%")) | |
| (Integration.description.ilike(f"%{search}%")) |
| ) |
| for i in query.all(): |
| integrations.append(IntegrationResponse( |
| id=i.id, name=i.name, description=i.description, logo_url=i.logo_url, |
| category=i.category, auth_type=i.auth_type, tool_count=i.tool_count, |
| trigger_count=i.trigger_count, is_active=i.is_active |
| )) |
| return integrations |
|
|
|
|
| @router.get("/categories") |
| async def list_categories(db: Session = Depends(get_db)): |
| categories = db.query(Integration.category).distinct().all() |
| return {"categories": list(set(c[0] for c in categories if c[0]))} |
|
|
|
|
| @router.get("/{integration_id}", response_model=IntegrationResponse) |
| async def get_integration(integration_id: str, db: Session = Depends(get_db)): |
| integration = db.query(Integration).filter(Integration.id == integration_id).first() |
| if not integration: |
| raise HTTPException(status_code=404, detail="Integration not found") |
| return IntegrationResponse( |
| id=integration.id, name=integration.name, description=integration.description, |
| logo_url=integration.logo_url, category=integration.category, |
| auth_type=integration.auth_type, tool_count=integration.tool_count, |
| trigger_count=integration.trigger_count, is_active=integration.is_active |
| ) |
|
|
|
|
| @router.post("/{integration_id}/connect") |
| async def connect_integration( |
| integration_id: str, |
| current_user: User = Depends(get_current_user), |
| db: Session = Depends(get_db) |
| ): |
| integration = db.query(Integration).filter(Integration.id == integration_id).first() |
| if not integration: |
| raise HTTPException(status_code=404, detail="Integration not found") |
| return { |
| "auth_url": f"/api/connections/oauth/begin?integration_id={integration_id}", |
| "message": "Use /api/connections/oauth/begin for OAuth flow" |
| } |
|
|
|
|
| @router.post("/sync") |
| async def sync_integrations( |
| force: bool = Query(False), |
| current_user: User = Depends(get_current_user), |
| db: Session = Depends(get_db) |
| ): |
| from integrations.registry import sync_from_composio |
| result = await sync_from_composio(force=force) |
| return {"message": "Integrations synced", "source": result["source"], "toolkits": result["toolkits"], "tools": result["tools"]} |
|
|
