Deploy emergent2api

Dockerfile

@@ -0,0 +1,17 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    gcc libpq-dev && \
+    rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt requirements.txt
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+ENV PORT=7860
+EXPOSE 7860
+
+CMD ["python", "-m", "emergent2api.app"]

README.md

@@ -0,0 +1,9 @@
+---
+title: Emergent2API
+emoji: 🤖
+colorFrom: blue
+colorTo: purple
+sdk: docker
+app_port: 7860
+pinned: false
+---

emergent2api/app.py

@@ -0,0 +1,185 @@
+"""Emergent2API — OpenAI/Anthropic-compatible API gateway for Emergent.sh accounts."""
+from __future__ import annotations
+
+import logging
+import os
+
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+
+from .config import settings
+from . import database as db
+from .pool import pool
+from .routes import openai, anthropic, responses
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(name)s] %(levelname)s %(message)s",
+)
+logger = logging.getLogger("emergent2api")
+
+app = FastAPI(
+    title="Emergent2API",
+    description="OpenAI/Anthropic-compatible API backed by Emergent.sh account pool",
+    version="0.1.0",
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+# ---------------------------------------------------------------------------
+# Auth middleware
+# ---------------------------------------------------------------------------
+
+@app.middleware("http")
+async def auth_middleware(request: Request, call_next):
+    path = request.url.path
+
+    # Skip auth for health check and docs
+    if path in ("/", "/health", "/docs", "/openapi.json", "/redoc"):
+        return await call_next(request)
+
+    # Check API key
+    auth_header = request.headers.get("authorization", "")
+    api_key = request.headers.get("x-api-key", "")
+
+    token = ""
+    if auth_header.startswith("Bearer "):
+        token = auth_header[7:]
+    elif api_key:
+        token = api_key
+
+    if token != settings.api_key:
+        return JSONResponse(
+            status_code=401,
+            content={"error": {"message": "Invalid API key", "type": "authentication_error"}},
+        )
+
+    return await call_next(request)
+
+
+# ---------------------------------------------------------------------------
+# Lifecycle
+# ---------------------------------------------------------------------------
+
+@app.on_event("startup")
+async def startup():
+    logger.info("Starting Emergent2API...")
+    logger.info(f"Backend: {settings.backend}")
+    logger.info(f"Proxy: {settings.proxy or 'none'}")
+
+    await db.init_db()
+    count = await pool.count()
+    logger.info(f"Account pool: {count['active']} active / {count['total']} total")
+
+
+@app.on_event("shutdown")
+async def shutdown():
+    await db.close_db()
+    logger.info("Emergent2API stopped")
+
+
+# ---------------------------------------------------------------------------
+# Routes
+# ---------------------------------------------------------------------------
+
+app.include_router(openai.router)
+app.include_router(anthropic.router)
+app.include_router(responses.router)
+
+
+@app.get("/")
+async def root():
+    count = await pool.count()
+    return {
+        "service": "Emergent2API",
+        "version": "0.1.0",
+        "backend": settings.backend,
+        "accounts": count,
+        "endpoints": [
+            "POST /v1/chat/completions (OpenAI)",
+            "POST /v1/messages (Anthropic)",
+            "POST /v1/responses (OpenAI Response API)",
+            "GET  /v1/models",
+            "GET  /admin/accounts",
+            "POST /admin/accounts",
+        ],
+    }
+
+
+@app.get("/health")
+async def health():
+    count = await pool.count()
+    return {"status": "ok", "accounts": count}
+
+
+# ---------------------------------------------------------------------------
+# Admin endpoints
+# ---------------------------------------------------------------------------
+
+@app.get("/admin/accounts")
+async def admin_list_accounts():
+    accounts = await db.get_active_accounts()
+    return {
+        "total": len(accounts),
+        "accounts": [
+            {
+                "id": a["id"],
+                "email": a["email"],
+                "balance": a["balance"],
+                "is_active": a["is_active"],
+                "last_used": a["last_used"].isoformat() if a["last_used"] else None,
+                "created_at": a["created_at"].isoformat() if a["created_at"] else None,
+            }
+            for a in accounts
+        ],
+    }
+
+
+@app.post("/admin/accounts")
+async def admin_add_account(request: Request):
+    """Add an account manually. Body: {email, password, jwt, refresh_token?, user_id?, balance?}"""
+    body = await request.json()
+    required = ["email", "password", "jwt"]
+    for field in required:
+        if field not in body:
+            return JSONResponse(
+                status_code=400,
+                content={"error": f"Missing required field: {field}"},
+            )
+    account_id = await db.upsert_account(body)
+    return {"id": account_id, "email": body["email"], "status": "added"}
+
+
+@app.post("/admin/accounts/import")
+async def admin_import_accounts(request: Request):
+    """Import accounts from JSONL. Body: {accounts: [{email, password, jwt, ...}, ...]}"""
+    body = await request.json()
+    accounts = body.get("accounts", [])
+    imported = 0
+    for acct in accounts:
+        if "email" in acct and "password" in acct and "jwt" in acct:
+            await db.upsert_account(acct)
+            imported += 1
+    return {"imported": imported, "total_submitted": len(accounts)}
+
+
+# ---------------------------------------------------------------------------
+# Entrypoint
+# ---------------------------------------------------------------------------
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(
+        "emergent2api.app:app",
+        host="0.0.0.0",
+        port=settings.port,
+        reload=False,
+    )

emergent2api/backends/base.py

@@ -0,0 +1,26 @@
+"""Abstract base for Emergent API backends."""
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any, AsyncGenerator, Optional
+
+
+class EmergentBackend(ABC):
+    @abstractmethod
+    async def chat(
+        self,
+        messages: list[dict[str, Any]],
+        model: str,
+        thinking: bool,
+        account: dict[str, Any],
+        stream: bool = False,
+    ) -> AsyncGenerator[dict[str, Any], None]:
+        """Yield response chunks. Each chunk is {"type": ..., "content": ...}.
+        
+        Chunk types:
+          - "thinking": reasoning content (extended thinking)
+          - "text": main response text
+          - "done": final signal with full text
+          - "error": error message
+        """
+        ...

emergent2api/backends/integrations.py

@@ -0,0 +1,100 @@
+"""Integrations API backend: proxy to integrations.emergentagent.com/llm."""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from typing import Any, AsyncGenerator
+
+import httpx
+
+from ..config import settings
+from .base import EmergentBackend
+
+logger = logging.getLogger("emergent2api.integrations")
+
+
+class IntegrationsBackend(EmergentBackend):
+    """Proxy to Emergent's OpenAI-compatible integrations endpoint.
+    
+    Requires a US-based proxy since the endpoint has IP restrictions.
+    """
+
+    async def chat(
+        self,
+        messages: list[dict[str, Any]],
+        model: str,
+        thinking: bool,
+        account: dict[str, Any],
+        stream: bool = False,
+    ) -> AsyncGenerator[dict[str, Any], None]:
+        jwt = account["jwt"]
+        url = f"{settings.integrations_base}/v1/chat/completions"
+
+        body = {
+            "model": model,
+            "messages": messages,
+            "stream": stream,
+        }
+
+        proxy = settings.proxy or None
+        transport = httpx.AsyncHTTPTransport(proxy=proxy) if proxy else None
+
+        try:
+            async with httpx.AsyncClient(
+                transport=transport,
+                timeout=httpx.Timeout(120.0, connect=15.0),
+            ) as client:
+                if stream:
+                    async with client.stream(
+                        "POST",
+                        url,
+                        json=body,
+                        headers={
+                            "Authorization": f"Bearer {jwt}",
+                            "Content-Type": "application/json",
+                        },
+                    ) as resp:
+                        if resp.status_code != 200:
+                            err = await resp.aread()
+                            yield {"type": "error", "content": f"Integrations API: HTTP {resp.status_code} {err.decode()[:200]}"}
+                            return
+
+                        full_text = ""
+                        async for line in resp.aiter_lines():
+                            if not line.startswith("data: "):
+                                continue
+                            data_str = line[6:].strip()
+                            if data_str == "[DONE]":
+                                break
+                            try:
+                                chunk = json.loads(data_str)
+                                delta = chunk.get("choices", [{}])[0].get("delta", {})
+                                content = delta.get("content", "")
+                                if content:
+                                    full_text += content
+                                    yield {"type": "text", "content": content}
+                            except json.JSONDecodeError:
+                                continue
+
+                        yield {"type": "done", "content": full_text, "thinking": ""}
+                else:
+                    resp = await client.post(
+                        url,
+                        json=body,
+                        headers={
+                            "Authorization": f"Bearer {jwt}",
+                            "Content-Type": "application/json",
+                        },
+                    )
+                    if resp.status_code != 200:
+                        yield {"type": "error", "content": f"Integrations API: HTTP {resp.status_code} {resp.text[:200]}"}
+                        return
+
+                    data = resp.json()
+                    content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
+                    yield {"type": "done", "content": content, "thinking": ""}
+
+        except Exception as e:
+            logger.error(f"Integrations backend error: {e}")
+            yield {"type": "error", "content": str(e)}

emergent2api/backends/jobs.py

@@ -0,0 +1,218 @@
+"""Jobs API backend: submit task + poll trajectory for responses."""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import time
+import uuid
+from typing import Any, AsyncGenerator
+
+from curl_cffi import requests as cffi_requests
+
+from ..config import settings
+from .base import EmergentBackend
+
+logger = logging.getLogger("emergent2api.jobs")
+
+
+class JobsBackend(EmergentBackend):
+    """Wraps Emergent's Jobs API (submit-queue + trajectory polling) into a chat interface."""
+
+    async def chat(
+        self,
+        messages: list[dict[str, Any]],
+        model: str,
+        thinking: bool,
+        account: dict[str, Any],
+        stream: bool = False,
+    ) -> AsyncGenerator[dict[str, Any], None]:
+        jwt = account["jwt"]
+
+        # Flatten messages into a single task prompt
+        task = self._messages_to_task(messages)
+
+        # Submit job
+        ref_id = await asyncio.to_thread(
+            self._submit_job, jwt, task, model, thinking
+        )
+        if not ref_id:
+            yield {"type": "error", "content": "Failed to submit job to Emergent"}
+            return
+
+        logger.info(f"Job submitted: ref_id={ref_id}, model={model}")
+
+        # Poll trajectory for results
+        full_text = ""
+        full_thinking = ""
+        seen_items: set[str] = set()
+        start_time = time.time()
+        empty_polls = 0
+
+        while (time.time() - start_time) < settings.poll_timeout:
+            await asyncio.sleep(settings.poll_interval)
+
+            result = await asyncio.to_thread(self._poll_trajectory, jwt, ref_id)
+            if result is None:
+                empty_polls += 1
+                if empty_polls > 5:
+                    continue
+                continue
+
+            data_items = result.get("data", [])
+            job_status = result.get("status", "")
+
+            new_content = False
+            for item in data_items:
+                item_id = str(item.get("id", item.get("request_id", "")))
+                if item_id in seen_items:
+                    continue
+                seen_items.add(item_id)
+
+                payload = item.get("traj_payload", {})
+                reasoning = payload.get("reasoning_content", "")
+                thought = payload.get("thought", "")
+
+                if reasoning and reasoning != full_thinking:
+                    delta = reasoning[len(full_thinking):]
+                    if delta:
+                        full_thinking = reasoning
+                        new_content = True
+                        if stream:
+                            yield {"type": "thinking", "content": delta}
+
+                if thought and thought != full_text:
+                    delta = thought[len(full_text):]
+                    if delta:
+                        full_text = thought
+                        new_content = True
+                        if stream:
+                            yield {"type": "text", "content": delta}
+
+            # Check for completion
+            if job_status in ("completed", "failed", "cancelled"):
+                break
+            if not new_content and data_items:
+                last_payload = data_items[-1].get("traj_payload", {})
+                if last_payload.get("status") in ("completed", "done"):
+                    break
+                final_thought = last_payload.get("thought", "")
+                if final_thought and final_thought == full_text and len(seen_items) > 1:
+                    # No new content after seeing items — likely done
+                    empty_polls += 1
+                    if empty_polls > 8:
+                        break
+
+        yield {
+            "type": "done",
+            "content": full_text,
+            "thinking": full_thinking,
+        }
+
+    def _submit_job(
+        self,
+        jwt: str,
+        task: str,
+        model_name: str,
+        thinking: bool,
+    ) -> str | None:
+        """Submit a job to Emergent's queue. Returns ref_id or None."""
+        s = self._make_session()
+        ref_id = str(uuid.uuid4())
+
+        conv_body = {
+            "client_ref_id": ref_id,
+            "payload": {
+                "processor_type": "env_only",
+                "is_cloud": True,
+                "env_image": "us-central1-docker.pkg.dev/emergent-default/emergent-container-hub/fastapi_react_mongo_shadcn_base_image_cloud_arm:release-26092025-2",
+                "branch": "",
+                "repository": "",
+                "enable_visual_edit": True,
+                "prompt_name": "auto_prompt_selector",
+                "prompt_version": "latest",
+                "work_space_dir": "",
+                "task": task,
+                "model_name": model_name,
+                "model_manually_selected": True,
+                "per_instance_cost_limit": 25,
+                "agentic_skills": [],
+                "plugin_version": "release-10092025-1",
+                "base64_image_list": [],
+                "human_timestamp": int(time.time() * 1000),
+                "asset_upload_enabled": True,
+                "is_pro_user": False,
+                "testMode": False,
+                "thinking_level": "thinking" if thinking else "none",
+                "job_mode": "public",
+                "mcp_id": [],
+            },
+            "model_name": model_name,
+            "resume": False,
+            "ads_metadata": {"app_version": settings.app_version},
+        }
+
+        try:
+            resp = s.post(
+                f"{settings.api_base}/jobs/v0/submit-queue/",
+                headers={
+                    "Authorization": f"Bearer {jwt}",
+                    "Origin": "https://app.emergent.sh",
+                    "Referer": "https://app.emergent.sh/",
+                },
+                json=conv_body,
+                timeout=30,
+            )
+            if resp.status_code in (200, 201, 202):
+                return ref_id
+            logger.error(f"Submit failed: HTTP {resp.status_code} {resp.text[:200]}")
+            return None
+        except Exception as e:
+            logger.error(f"Submit error: {e}")
+            return None
+
+    def _poll_trajectory(self, jwt: str, ref_id: str) -> dict | None:
+        """Poll trajectory history for a given ref_id."""
+        s = self._make_session()
+        try:
+            resp = s.get(
+                f"{settings.api_base}/trajectories/v0/{ref_id}/history?limit=50",
+                headers={"Authorization": f"Bearer {jwt}"},
+                timeout=15,
+            )
+            if resp.status_code == 200:
+                return resp.json()
+            return None
+        except Exception:
+            return None
+
+    def _make_session(self) -> cffi_requests.Session:
+        s = cffi_requests.Session(impersonate="chrome")
+        if settings.proxy:
+            s.proxies = {"http": settings.proxy, "https": settings.proxy}
+        return s
+
+    @staticmethod
+    def _messages_to_task(messages: list[dict[str, Any]]) -> str:
+        """Convert chat messages into a single task string for Emergent."""
+        parts = []
+        for msg in messages:
+            role = msg.get("role", "user")
+            content = msg.get("content", "")
+            if isinstance(content, list):
+                text_parts = []
+                for item in content:
+                    if isinstance(item, dict) and item.get("type") == "text":
+                        text_parts.append(item.get("text", ""))
+                    elif isinstance(item, str):
+                        text_parts.append(item)
+                content = "\n".join(text_parts)
+
+            if role == "system":
+                parts.append(f"[System Instructions]\n{content}")
+            elif role == "assistant":
+                parts.append(f"[Previous Assistant Response]\n{content}")
+            else:
+                parts.append(content)
+
+        return "\n\n".join(parts)

emergent2api/config.py

@@ -0,0 +1,34 @@
+"""Configuration for emergent2api service."""
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, field
+
+
+@dataclass
+class Settings:
+    api_key: str = field(default_factory=lambda: os.environ.get("EMERGENT2API_KEY", "sk-emergent2api"))
+    backend: str = field(default_factory=lambda: os.environ.get("EMERGENT_BACKEND", "jobs"))
+    proxy: str = field(default_factory=lambda: os.environ.get("EMERGENT_PROXY", ""))
+    poll_interval: float = field(default_factory=lambda: float(os.environ.get("EMERGENT_POLL_INTERVAL", "0.8")))
+    poll_timeout: float = field(default_factory=lambda: float(os.environ.get("EMERGENT_POLL_TIMEOUT", "120")))
+    database_url: str = field(default_factory=lambda: os.environ.get(
+        "DATABASE_URL",
+        "postgresql://neondb_owner:npg_cyGIuK58kePT@ep-muddy-sky-adych24h-pooler.c-2.us-east-1.aws.neon.tech/neondb?sslmode=require",
+    ))
+    port: int = field(default_factory=lambda: int(os.environ.get("PORT", "8000")))
+
+    # Supabase constants
+    supabase_anon_key: str = (
+        "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
+        "eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6InNua3N4d2t5dW1oZHlreXJoaGNoIiwi"
+        "cm9sZSI6ImFub24iLCJpYXQiOjE3MjQ3NzI2NDYsImV4cCI6MjA0MDM0ODY0Nn0."
+        "3unO6zdz2NilPL2xdxt7OjvZA19copj3Q7ulIjPVDLQ"
+    )
+    auth_base: str = "https://auth.emergent.sh/auth/v1"
+    api_base: str = "https://api.emergent.sh"
+    integrations_base: str = "https://integrations.emergentagent.com/llm"
+    app_version: str = "1.1.28"
+
+
+settings = Settings()

emergent2api/database.py

@@ -0,0 +1,126 @@
+"""PostgreSQL database layer for account pool management."""
+from __future__ import annotations
+
+import asyncio
+import json
+import time
+from datetime import datetime, timezone
+from typing import Any, Optional
+
+import asyncpg
+
+from .config import settings
+
+_pool: Optional[asyncpg.Pool] = None
+
+SCHEMA_SQL = """
+CREATE TABLE IF NOT EXISTS emergent_accounts (
+    id          SERIAL PRIMARY KEY,
+    email       TEXT NOT NULL UNIQUE,
+    password    TEXT NOT NULL,
+    jwt         TEXT NOT NULL,
+    refresh_token TEXT DEFAULT '',
+    user_id     TEXT DEFAULT '',
+    balance     REAL DEFAULT 0,
+    is_active   BOOLEAN DEFAULT TRUE,
+    last_used   TIMESTAMPTZ,
+    created_at  TIMESTAMPTZ DEFAULT NOW(),
+    updated_at  TIMESTAMPTZ DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS idx_emergent_active ON emergent_accounts (is_active) WHERE is_active = TRUE;
+"""
+
+
+async def get_pool() -> asyncpg.Pool:
+    global _pool
+    if _pool is None:
+        _pool = await asyncpg.create_pool(
+            settings.database_url,
+            min_size=2,
+            max_size=10,
+            statement_cache_size=0,
+        )
+    return _pool
+
+
+async def init_db() -> None:
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        await conn.execute(SCHEMA_SQL)
+
+
+async def close_db() -> None:
+    global _pool
+    if _pool:
+        await _pool.close()
+        _pool = None
+
+
+async def upsert_account(account: dict[str, Any]) -> int:
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(
+            """
+            INSERT INTO emergent_accounts (email, password, jwt, refresh_token, user_id, balance)
+            VALUES ($1, $2, $3, $4, $5, $6)
+            ON CONFLICT (email) DO UPDATE SET
+                jwt = EXCLUDED.jwt,
+                refresh_token = EXCLUDED.refresh_token,
+                balance = EXCLUDED.balance,
+                is_active = TRUE,
+                updated_at = NOW()
+            RETURNING id
+            """,
+            account["email"],
+            account["password"],
+            account["jwt"],
+            account.get("refresh_token", ""),
+            account.get("user_id", ""),
+            account.get("balance", 0),
+        )
+        return row["id"]
+
+
+async def get_active_accounts() -> list[dict[str, Any]]:
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        rows = await conn.fetch(
+            "SELECT * FROM emergent_accounts WHERE is_active = TRUE ORDER BY last_used ASC NULLS FIRST"
+        )
+        return [dict(r) for r in rows]
+
+
+async def mark_used(account_id: int) -> None:
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        await conn.execute(
+            "UPDATE emergent_accounts SET last_used = NOW() WHERE id = $1",
+            account_id,
+        )
+
+
+async def deactivate_account(account_id: int) -> None:
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        await conn.execute(
+            "UPDATE emergent_accounts SET is_active = FALSE, updated_at = NOW() WHERE id = $1",
+            account_id,
+        )
+
+
+async def update_jwt(account_id: int, jwt: str, refresh_token: str = "") -> None:
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        await conn.execute(
+            "UPDATE emergent_accounts SET jwt = $2, refresh_token = $3, updated_at = NOW() WHERE id = $1",
+            account_id, jwt, refresh_token,
+        )
+
+
+async def get_account_count() -> dict[str, int]:
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        total = await conn.fetchval("SELECT COUNT(*) FROM emergent_accounts")
+        active = await conn.fetchval("SELECT COUNT(*) FROM emergent_accounts WHERE is_active = TRUE")
+        return {"total": total, "active": active}

emergent2api/models.py

@@ -0,0 +1,161 @@
+"""Pydantic models for OpenAI, Anthropic, and Response API formats."""
+from __future__ import annotations
+
+import time
+import uuid
+from typing import Any, Optional, Union
+
+from pydantic import BaseModel, Field
+
+# ---------------------------------------------------------------------------
+# Model mapping: user-facing name -> Emergent internal name
+# ---------------------------------------------------------------------------
+
+MODEL_MAP = {
+    "claude-opus-4-6": "claude-opus-4-6",
+    "claude-opus-4.6": "claude-opus-4-6",
+    "claude-sonnet-4-5": "claude-sonnet-4-5",
+    "claude-sonnet-4.5": "claude-sonnet-4-5",
+    "claude-sonnet-4-5-thinking": "claude-sonnet-4-5",
+    "claude-sonnet-4.5-thinking": "claude-sonnet-4-5",
+}
+
+AVAILABLE_MODELS = [
+    {"id": "claude-opus-4-6", "object": "model", "created": 1709251200, "owned_by": "emergent"},
+    {"id": "claude-sonnet-4-5", "object": "model", "created": 1709251200, "owned_by": "emergent"},
+    {"id": "claude-sonnet-4-5-thinking", "object": "model", "created": 1709251200, "owned_by": "emergent"},
+]
+
+
+def resolve_model(model: str) -> tuple[str, bool]:
+    """Resolve user model name to (emergent_model, thinking_enabled)."""
+    thinking = "thinking" in model.lower()
+    emergent_model = MODEL_MAP.get(model, model)
+    if emergent_model not in ("claude-opus-4-6", "claude-sonnet-4-5"):
+        emergent_model = "claude-sonnet-4-5"
+    return emergent_model, thinking
+
+
+# ---------------------------------------------------------------------------
+# OpenAI Chat Completions
+# ---------------------------------------------------------------------------
+
+class OAIMessage(BaseModel):
+    role: str
+    content: Union[str, list[Any]]
+    name: Optional[str] = None
+
+
+class OAIChatRequest(BaseModel):
+    model: str = "claude-sonnet-4-5"
+    messages: list[OAIMessage]
+    stream: bool = False
+    temperature: Optional[float] = None
+    max_tokens: Optional[int] = None
+    top_p: Optional[float] = None
+    user: Optional[str] = None
+
+
+class OAIChatChoice(BaseModel):
+    index: int = 0
+    message: dict[str, Any]
+    finish_reason: Optional[str] = "stop"
+
+
+class OAIChatResponse(BaseModel):
+    id: str = Field(default_factory=lambda: f"chatcmpl-{uuid.uuid4().hex[:24]}")
+    object: str = "chat.completion"
+    created: int = Field(default_factory=lambda: int(time.time()))
+    model: str = ""
+    choices: list[OAIChatChoice] = []
+    usage: dict[str, int] = Field(default_factory=lambda: {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0})
+
+
+class OAIStreamDelta(BaseModel):
+    role: Optional[str] = None
+    content: Optional[str] = None
+
+
+class OAIStreamChoice(BaseModel):
+    index: int = 0
+    delta: OAIStreamDelta
+    finish_reason: Optional[str] = None
+
+
+class OAIStreamChunk(BaseModel):
+    id: str = ""
+    object: str = "chat.completion.chunk"
+    created: int = 0
+    model: str = ""
+    choices: list[OAIStreamChoice] = []
+
+
+# ---------------------------------------------------------------------------
+# Anthropic Messages API
+# ---------------------------------------------------------------------------
+
+class AnthropicMessage(BaseModel):
+    role: str
+    content: Union[str, list[Any]]
+
+
+class AnthropicRequest(BaseModel):
+    model: str = "claude-sonnet-4-5"
+    messages: list[AnthropicMessage]
+    max_tokens: int = 4096
+    stream: bool = False
+    temperature: Optional[float] = None
+    top_p: Optional[float] = None
+    system: Optional[str] = None
+
+
+class AnthropicContentBlock(BaseModel):
+    type: str = "text"
+    text: str = ""
+
+
+class AnthropicResponse(BaseModel):
+    id: str = Field(default_factory=lambda: f"msg_{uuid.uuid4().hex[:24]}")
+    type: str = "message"
+    role: str = "assistant"
+    content: list[AnthropicContentBlock] = []
+    model: str = ""
+    stop_reason: Optional[str] = "end_turn"
+    stop_sequence: Optional[str] = None
+    usage: dict[str, int] = Field(default_factory=lambda: {"input_tokens": 0, "output_tokens": 0})
+
+
+# ---------------------------------------------------------------------------
+# OpenAI Response API
+# ---------------------------------------------------------------------------
+
+class ResponseAPIInput(BaseModel):
+    role: str = "user"
+    content: Union[str, list[Any]] = ""
+
+
+class ResponseAPIRequest(BaseModel):
+    model: str = "claude-sonnet-4-5"
+    input: Union[str, list[ResponseAPIInput]] = ""
+    stream: bool = False
+    instructions: Optional[str] = None
+    max_output_tokens: Optional[int] = None
+    temperature: Optional[float] = None
+
+
+class ResponseAPIOutput(BaseModel):
+    type: str = "message"
+    id: str = Field(default_factory=lambda: f"msg_{uuid.uuid4().hex[:16]}")
+    status: str = "completed"
+    role: str = "assistant"
+    content: list[dict[str, Any]] = Field(default_factory=list)
+
+
+class ResponseAPIResponse(BaseModel):
+    id: str = Field(default_factory=lambda: f"resp_{uuid.uuid4().hex[:24]}")
+    object: str = "response"
+    created_at: int = Field(default_factory=lambda: int(time.time()))
+    status: str = "completed"
+    model: str = ""
+    output: list[ResponseAPIOutput] = []
+    usage: dict[str, int] = Field(default_factory=lambda: {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0})

emergent2api/pool.py

@@ -0,0 +1,118 @@
+"""Account pool manager with round-robin selection and JWT refresh."""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import time
+from typing import Any, Optional
+
+from curl_cffi import requests as cffi_requests
+
+from .config import settings
+from . import database as db
+
+logger = logging.getLogger("emergent2api.pool")
+
+
+class AccountPool:
+    """Thread-safe round-robin account pool backed by PostgreSQL."""
+
+    def __init__(self):
+        self._index = 0
+        self._lock = asyncio.Lock()
+        self._cache: list[dict[str, Any]] = []
+        self._cache_ts: float = 0
+        self._cache_ttl: float = 30  # refresh cache every 30s
+
+    async def _refresh_cache(self, force: bool = False) -> None:
+        now = time.time()
+        if not force and self._cache and (now - self._cache_ts) < self._cache_ttl:
+            return
+        self._cache = await db.get_active_accounts()
+        self._cache_ts = now
+        logger.info(f"Pool cache refreshed: {len(self._cache)} active accounts")
+
+    async def acquire(self) -> Optional[dict[str, Any]]:
+        """Get next account via round-robin. Returns account dict or None."""
+        async with self._lock:
+            await self._refresh_cache()
+            if not self._cache:
+                return None
+            account = self._cache[self._index % len(self._cache)]
+            self._index = (self._index + 1) % len(self._cache)
+        await db.mark_used(account["id"])
+        return account
+
+    async def release_bad(self, account_id: int) -> None:
+        """Mark an account as inactive after auth failure."""
+        await db.deactivate_account(account_id)
+        async with self._lock:
+            await self._refresh_cache(force=True)
+        logger.warning(f"Account {account_id} deactivated")
+
+    async def refresh_jwt(self, account: dict[str, Any]) -> Optional[str]:
+        """Re-login to get a fresh JWT using stored email/password."""
+        try:
+            s = cffi_requests.Session(impersonate="chrome")
+            if settings.proxy:
+                s.proxies = {"http": settings.proxy, "https": settings.proxy}
+
+            resp = s.post(
+                f"{settings.auth_base}/token?grant_type=password",
+                json={
+                    "email": account["email"],
+                    "password": account["password"],
+                    "gotrue_meta_security": {},
+                },
+                headers={
+                    "Apikey": settings.supabase_anon_key,
+                    "Authorization": f"Bearer {settings.supabase_anon_key}",
+                    "X-Client-Info": "supabase-js-web/2.58.0",
+                },
+                timeout=15,
+            )
+            if resp.status_code != 200:
+                logger.error(f"JWT refresh failed for {account['email']}: HTTP {resp.status_code}")
+                return None
+
+            data = resp.json()
+            new_jwt = data["access_token"]
+            new_refresh = data.get("refresh_token", "")
+            await db.update_jwt(account["id"], new_jwt, new_refresh)
+            logger.info(f"JWT refreshed for {account['email']}")
+
+            # Re-initialize user details (mandatory)
+            self._init_user_detail(s, new_jwt)
+
+            return new_jwt
+        except Exception as e:
+            logger.error(f"JWT refresh error for {account['email']}: {e}")
+            return None
+
+    @staticmethod
+    def _init_user_detail(s: cffi_requests.Session, jwt: str) -> None:
+        """Call get_user_detail — mandatory for account activation."""
+        try:
+            s.post(
+                f"{settings.api_base}/user/details",
+                data=json.dumps({
+                    "ads_metadata": {
+                        "app_version": settings.app_version,
+                        "showError": "You are unauthorized to use this application",
+                    }
+                }),
+                headers={
+                    "Content-Type": "application/json",
+                    "Authorization": f"Bearer {jwt}",
+                },
+                timeout=15,
+            )
+        except Exception:
+            pass
+
+    async def count(self) -> dict[str, int]:
+        return await db.get_account_count()
+
+
+pool = AccountPool()

emergent2api/routes/anthropic.py

@@ -0,0 +1,121 @@
+"""Anthropic-compatible /v1/messages endpoint."""
+from __future__ import annotations
+
+import json
+import time
+import uuid
+
+from fastapi import APIRouter, Request
+from fastapi.responses import JSONResponse, StreamingResponse
+
+from ..models import (
+    AnthropicRequest,
+    AnthropicResponse,
+    AnthropicContentBlock,
+    resolve_model,
+)
+from ..pool import pool
+from ..backends.jobs import JobsBackend
+from ..backends.integrations import IntegrationsBackend
+from ..config import settings
+
+router = APIRouter()
+
+
+def _get_backend():
+    if settings.backend == "integrations":
+        return IntegrationsBackend()
+    return JobsBackend()
+
+
+@router.post("/v1/messages")
+async def messages(request: Request):
+    body = await request.json()
+    req = AnthropicRequest(**body)
+
+    account = await pool.acquire()
+    if not account:
+        return JSONResponse(
+            status_code=503,
+            content={"type": "error", "error": {"type": "overloaded_error", "message": "No active accounts available"}},
+        )
+
+    emergent_model, thinking = resolve_model(req.model)
+    backend = _get_backend()
+
+    # Build messages list (prepend system if present)
+    messages = []
+    if req.system:
+        messages.append({"role": "system", "content": req.system})
+    for m in req.messages:
+        messages.append({"role": m.role, "content": m.content})
+
+    if req.stream:
+        return StreamingResponse(
+            _stream_anthropic(backend, messages, emergent_model, thinking, account, req.model),
+            media_type="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+                "X-Accel-Buffering": "no",
+            },
+        )
+
+    # Non-streaming
+    full_text = ""
+    error_msg = ""
+    async for chunk in backend.chat(messages, emergent_model, thinking, account, stream=False):
+        if chunk["type"] == "error":
+            error_msg = chunk["content"]
+            break
+        if chunk["type"] == "done":
+            full_text = chunk["content"]
+
+    if error_msg:
+        return JSONResponse(
+            status_code=500,
+            content={"type": "error", "error": {"type": "api_error", "message": error_msg}},
+        )
+
+    response = AnthropicResponse(
+        model=req.model,
+        content=[AnthropicContentBlock(text=full_text)],
+    )
+    return response.model_dump()
+
+
+async def _stream_anthropic(backend, messages, emergent_model, thinking, account, user_model):
+    """Generate SSE stream in Anthropic Messages format."""
+    msg_id = f"msg_{uuid.uuid4().hex[:24]}"
+
+    # message_start
+    yield f"event: message_start\ndata: {json.dumps({'type': 'message_start', 'message': {'id': msg_id, 'type': 'message', 'role': 'assistant', 'content': [], 'model': user_model, 'stop_reason': None, 'stop_sequence': None, 'usage': {'input_tokens': 0, 'output_tokens': 0}}})}\n\n"
+
+    # content_block_start
+    yield f"event: content_block_start\ndata: {json.dumps({'type': 'content_block_start', 'index': 0, 'content_block': {'type': 'text', 'text': ''}})}\n\n"
+
+    total_output = 0
+    async for chunk in backend.chat(messages, emergent_model, thinking, account, stream=True):
+        if chunk["type"] == "error":
+            err_text = chunk["content"]
+            yield f"event: content_block_delta\ndata: {json.dumps({'type': 'content_block_delta', 'index': 0, 'delta': {'type': 'text_delta', 'text': f'[Error: {err_text}]'}})}\n\n"
+            break
+
+        if chunk["type"] in ("text", "thinking"):
+            content = chunk["content"]
+            if not content:
+                continue
+            total_output += len(content)
+            yield f"event: content_block_delta\ndata: {json.dumps({'type': 'content_block_delta', 'index': 0, 'delta': {'type': 'text_delta', 'text': content}})}\n\n"
+
+        if chunk["type"] == "done":
+            break
+
+    # content_block_stop
+    yield f"event: content_block_stop\ndata: {json.dumps({'type': 'content_block_stop', 'index': 0})}\n\n"
+
+    # message_delta
+    yield f"event: message_delta\ndata: {json.dumps({'type': 'message_delta', 'delta': {'stop_reason': 'end_turn', 'stop_sequence': None}, 'usage': {'output_tokens': total_output}})}\n\n"
+
+    # message_stop
+    yield f"event: message_stop\ndata: {json.dumps({'type': 'message_stop'})}\n\n"

emergent2api/routes/openai.py

@@ -0,0 +1,148 @@
+"""OpenAI-compatible /v1/chat/completions and /v1/models endpoints."""
+from __future__ import annotations
+
+import json
+import time
+import uuid
+
+from fastapi import APIRouter, Request
+from fastapi.responses import JSONResponse, StreamingResponse
+
+from ..models import (
+    AVAILABLE_MODELS,
+    OAIChatRequest,
+    OAIChatResponse,
+    OAIChatChoice,
+    OAIStreamChunk,
+    OAIStreamChoice,
+    OAIStreamDelta,
+    resolve_model,
+)
+from ..pool import pool
+from ..backends.jobs import JobsBackend
+from ..backends.integrations import IntegrationsBackend
+from ..config import settings
+
+router = APIRouter()
+
+def _get_backend():
+    if settings.backend == "integrations":
+        return IntegrationsBackend()
+    return JobsBackend()
+
+
+@router.get("/v1/models")
+async def list_models():
+    return {"object": "list", "data": AVAILABLE_MODELS}
+
+
+@router.post("/v1/chat/completions")
+async def chat_completions(request: Request):
+    body = await request.json()
+    req = OAIChatRequest(**body)
+
+    account = await pool.acquire()
+    if not account:
+        return JSONResponse(
+            status_code=503,
+            content={"error": {"message": "No active accounts available", "type": "server_error"}},
+        )
+
+    emergent_model, thinking = resolve_model(req.model)
+    backend = _get_backend()
+    messages = [{"role": m.role, "content": m.content} for m in req.messages]
+
+    if req.stream:
+        return StreamingResponse(
+            _stream_response(backend, messages, emergent_model, thinking, account, req.model),
+            media_type="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+                "X-Accel-Buffering": "no",
+            },
+        )
+
+    # Non-streaming
+    full_text = ""
+    full_thinking = ""
+    error_msg = ""
+    async for chunk in backend.chat(messages, emergent_model, thinking, account, stream=False):
+        if chunk["type"] == "error":
+            error_msg = chunk["content"]
+            break
+        if chunk["type"] == "done":
+            full_text = chunk["content"]
+            full_thinking = chunk.get("thinking", "")
+
+    if error_msg:
+        return JSONResponse(
+            status_code=500,
+            content={"error": {"message": error_msg, "type": "server_error"}},
+        )
+
+    response = OAIChatResponse(
+        model=req.model,
+        choices=[OAIChatChoice(
+            message={"role": "assistant", "content": full_text},
+            finish_reason="stop",
+        )],
+    )
+    return response.model_dump()
+
+
+async def _stream_response(backend, messages, emergent_model, thinking, account, user_model):
+    """Generate SSE stream in OpenAI chat completions format."""
+    chat_id = f"chatcmpl-{uuid.uuid4().hex[:24]}"
+    created = int(time.time())
+
+    # Initial chunk with role
+    initial = OAIStreamChunk(
+        id=chat_id,
+        created=created,
+        model=user_model,
+        choices=[OAIStreamChoice(
+            delta=OAIStreamDelta(role="assistant"),
+        )],
+    )
+    yield f"data: {json.dumps(initial.model_dump())}\n\n"
+
+    async for chunk in backend.chat(messages, emergent_model, thinking, account, stream=True):
+        if chunk["type"] == "error":
+            error_chunk = {
+                "id": chat_id, "object": "chat.completion.chunk",
+                "created": created, "model": user_model,
+                "choices": [{"index": 0, "delta": {"content": f"\n[Error: {chunk['content']}]"}, "finish_reason": None}],
+            }
+            yield f"data: {json.dumps(error_chunk)}\n\n"
+            break
+
+        if chunk["type"] in ("text", "thinking"):
+            content = chunk["content"]
+            if not content:
+                continue
+            stream_chunk = OAIStreamChunk(
+                id=chat_id,
+                created=created,
+                model=user_model,
+                choices=[OAIStreamChoice(
+                    delta=OAIStreamDelta(content=content),
+                )],
+            )
+            yield f"data: {json.dumps(stream_chunk.model_dump())}\n\n"
+
+        if chunk["type"] == "done":
+            break
+
+    # Final chunk
+    final = OAIStreamChunk(
+        id=chat_id,
+        created=created,
+        model=user_model,
+        choices=[OAIStreamChoice(
+            delta=OAIStreamDelta(),
+            finish_reason="stop",
+        )],
+    )
+    yield f"data: {json.dumps(final.model_dump())}\n\n"
+    yield "data: [DONE]\n\n"

emergent2api/routes/responses.py

@@ -0,0 +1,138 @@
+"""OpenAI Response API /v1/responses endpoint."""
+from __future__ import annotations
+
+import json
+import time
+import uuid
+
+from fastapi import APIRouter, Request
+from fastapi.responses import JSONResponse, StreamingResponse
+
+from ..models import (
+    ResponseAPIRequest,
+    ResponseAPIResponse,
+    ResponseAPIOutput,
+    resolve_model,
+)
+from ..pool import pool
+from ..backends.jobs import JobsBackend
+from ..backends.integrations import IntegrationsBackend
+from ..config import settings
+
+router = APIRouter()
+
+
+def _get_backend():
+    if settings.backend == "integrations":
+        return IntegrationsBackend()
+    return JobsBackend()
+
+
+def _input_to_messages(req: ResponseAPIRequest) -> list[dict]:
+    """Convert Response API input format to chat messages."""
+    messages = []
+    if req.instructions:
+        messages.append({"role": "system", "content": req.instructions})
+
+    if isinstance(req.input, str):
+        messages.append({"role": "user", "content": req.input})
+    elif isinstance(req.input, list):
+        for item in req.input:
+            messages.append({"role": item.role, "content": item.content})
+
+    return messages
+
+
+@router.post("/v1/responses")
+async def responses(request: Request):
+    body = await request.json()
+    req = ResponseAPIRequest(**body)
+
+    account = await pool.acquire()
+    if not account:
+        return JSONResponse(
+            status_code=503,
+            content={"error": {"message": "No active accounts available", "type": "server_error"}},
+        )
+
+    emergent_model, thinking = resolve_model(req.model)
+    backend = _get_backend()
+    messages = _input_to_messages(req)
+
+    if req.stream:
+        return StreamingResponse(
+            _stream_responses(backend, messages, emergent_model, thinking, account, req.model),
+            media_type="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+                "X-Accel-Buffering": "no",
+            },
+        )
+
+    # Non-streaming
+    full_text = ""
+    error_msg = ""
+    async for chunk in backend.chat(messages, emergent_model, thinking, account, stream=False):
+        if chunk["type"] == "error":
+            error_msg = chunk["content"]
+            break
+        if chunk["type"] == "done":
+            full_text = chunk["content"]
+
+    if error_msg:
+        return JSONResponse(
+            status_code=500,
+            content={"error": {"message": error_msg, "type": "server_error"}},
+        )
+
+    resp_id = f"resp_{uuid.uuid4().hex[:24]}"
+    output_item = ResponseAPIOutput(
+        content=[{"type": "output_text", "text": full_text}],
+    )
+    response = ResponseAPIResponse(
+        id=resp_id,
+        model=req.model,
+        output=[output_item],
+    )
+    return response.model_dump()
+
+
+async def _stream_responses(backend, messages, emergent_model, thinking, account, user_model):
+    """Generate SSE stream in OpenAI Response API format."""
+    resp_id = f"resp_{uuid.uuid4().hex[:24]}"
+    item_id = f"msg_{uuid.uuid4().hex[:16]}"
+    output_idx = 0
+
+    # response.created
+    yield f"event: response.created\ndata: {json.dumps({'type': 'response.created', 'response': {'id': resp_id, 'object': 'response', 'status': 'in_progress', 'model': user_model}})}\n\n"
+
+    # response.output_item.added
+    yield f"event: response.output_item.added\ndata: {json.dumps({'type': 'response.output_item.added', 'output_index': output_idx, 'item': {'type': 'message', 'id': item_id, 'role': 'assistant', 'status': 'in_progress', 'content': []}})}\n\n"
+
+    # response.content_part.added
+    yield f"event: response.content_part.added\ndata: {json.dumps({'type': 'response.content_part.added', 'item_id': item_id, 'output_index': output_idx, 'content_index': 0, 'part': {'type': 'output_text', 'text': ''}})}\n\n"
+
+    async for chunk in backend.chat(messages, emergent_model, thinking, account, stream=True):
+        if chunk["type"] == "error":
+            err_text = chunk["content"]
+            yield f"event: response.output_text.delta\ndata: {json.dumps({'type': 'response.output_text.delta', 'item_id': item_id, 'output_index': output_idx, 'content_index': 0, 'delta': f'[Error: {err_text}]'})}\n\n"
+            break
+
+        if chunk["type"] in ("text", "thinking"):
+            content = chunk["content"]
+            if not content:
+                continue
+            yield f"event: response.output_text.delta\ndata: {json.dumps({'type': 'response.output_text.delta', 'item_id': item_id, 'output_index': output_idx, 'content_index': 0, 'delta': content})}\n\n"
+
+        if chunk["type"] == "done":
+            break
+
+    # response.content_part.done
+    yield f"event: response.content_part.done\ndata: {json.dumps({'type': 'response.content_part.done', 'item_id': item_id, 'output_index': output_idx, 'content_index': 0})}\n\n"
+
+    # response.output_item.done
+    yield f"event: response.output_item.done\ndata: {json.dumps({'type': 'response.output_item.done', 'output_index': output_idx, 'item': {'type': 'message', 'id': item_id, 'role': 'assistant', 'status': 'completed'}})}\n\n"
+
+    # response.completed
+    yield f"event: response.completed\ndata: {json.dumps({'type': 'response.completed', 'response': {'id': resp_id, 'object': 'response', 'status': 'completed', 'model': user_model}})}\n\n"

requirements.txt

@@ -0,0 +1,6 @@
+fastapi>=0.115
+uvicorn[standard]>=0.30
+pydantic-settings>=2.0
+asyncpg>=0.29
+httpx>=0.27
+curl_cffi>=0.7