Update emergent2api v0.2.0: auth_token, admin endpoints, account management

- Default auth_token: sk-6loA0HwMQP1mdhPvI
- Add JSONL import, toggle, delete, JWT refresh endpoints
- Update app_version to 3.5.0
- Add python-multipart dependency

Made-with: Cursor

emergent2api/__init__.py

@@ -0,0 +1 @@
+

emergent2api/app.py

@@ -1,6 +1,7 @@
 """Emergent2API — OpenAI/Anthropic-compatible API gateway for Emergent.sh accounts."""
 from __future__ import annotations
 
+import json
 import logging
 import os
 
@@ -100,17 +101,27 @@ async def root():
     count = await pool.count()
     return {
         "service": "Emergent2API",
-        "version": "0.1.0",
+        "version": "0.2.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",
-        ],
+        "endpoints": {
+            "api": [
+                "POST /v1/chat/completions (OpenAI)",
+                "POST /v1/messages (Anthropic)",
+                "POST /v1/responses (OpenAI Response API)",
+                "GET  /v1/models",
+            ],
+            "admin": [
+                "GET    /admin/accounts",
+                "POST   /admin/accounts",
+                "POST   /admin/accounts/import",
+                "POST   /admin/accounts/import-jsonl",
+                "DELETE /admin/accounts/{id}",
+                "POST   /admin/accounts/{id}/toggle",
+                "POST   /admin/accounts/{id}/refresh",
+                "POST   /admin/accounts/refresh-all",
+            ],
+        },
     }
 
 
@@ -125,8 +136,12 @@ async def health():
 # ---------------------------------------------------------------------------
 
 @app.get("/admin/accounts")
-async def admin_list_accounts():
-    accounts = await db.get_active_accounts()
+async def admin_list_accounts(active_only: bool = False):
+    """List all accounts or only active ones."""
+    if active_only:
+        accounts = await db.get_active_accounts()
+    else:
+        accounts = await db.get_all_accounts()
     return {
         "total": len(accounts),
         "accounts": [
@@ -148,27 +163,101 @@ 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}"},
-            )
+    for f in required:
+        if f not in body:
+            return JSONResponse(status_code=400, content={"error": f"Missing required field: {f}"})
     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, ...}, ...]}"""
+    """Import accounts from JSON body. Body: {accounts: [{email, password, jwt, ...}, ...]}"""
     body = await request.json()
     accounts = body.get("accounts", [])
     imported = 0
+    errors = []
     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)}
+            try:
+                await db.upsert_account(acct)
+                imported += 1
+            except Exception as e:
+                errors.append({"email": acct.get("email"), "error": str(e)})
+        else:
+            errors.append({"email": acct.get("email", "?"), "error": "missing email/password/jwt"})
+    return {"imported": imported, "total_submitted": len(accounts), "errors": errors}
+
+
+@app.post("/admin/accounts/import-jsonl")
+async def admin_import_jsonl(request: Request):
+    """Import accounts from JSONL text. Body: {jsonl: "line1\\nline2\\n..."} or raw JSONL body."""
+    content_type = request.headers.get("content-type", "")
+    if "application/json" in content_type:
+        body = await request.json()
+        raw_text = body.get("jsonl", "")
+    else:
+        raw_bytes = await request.body()
+        raw_text = raw_bytes.decode("utf-8", errors="replace")
+    lines = raw_text.strip().splitlines()
+    imported = 0
+    errors = []
+    for i, line in enumerate(lines):
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            acct = json.loads(line)
+            if "email" in acct and "password" in acct and "jwt" in acct:
+                await db.upsert_account(acct)
+                imported += 1
+            else:
+                errors.append({"line": i + 1, "error": "missing email/password/jwt"})
+        except json.JSONDecodeError:
+            errors.append({"line": i + 1, "error": "invalid JSON"})
+        except Exception as e:
+            errors.append({"line": i + 1, "error": str(e)})
+    return {"imported": imported, "total_lines": len(lines), "errors": errors}
+
+
+@app.delete("/admin/accounts/{account_id}")
+async def admin_delete_account(account_id: int):
+    """Permanently delete an account."""
+    await db.delete_account(account_id)
+    return {"id": account_id, "status": "deleted"}
+
+
+@app.post("/admin/accounts/{account_id}/toggle")
+async def admin_toggle_account(account_id: int):
+    """Toggle active/inactive status."""
+    new_state = await db.toggle_account(account_id)
+    return {"id": account_id, "is_active": new_state}
+
+
+@app.post("/admin/accounts/{account_id}/refresh")
+async def admin_refresh_jwt(account_id: int):
+    """Refresh JWT for a specific account."""
+    account = await db.get_account_by_id(account_id)
+    if not account:
+        return JSONResponse(status_code=404, content={"error": "Account not found"})
+    new_jwt = await pool.refresh_jwt(account)
+    if new_jwt:
+        return {"id": account_id, "email": account["email"], "status": "refreshed"}
+    return JSONResponse(status_code=500, content={"error": "JWT refresh failed"})
+
+
+@app.post("/admin/accounts/refresh-all")
+async def admin_refresh_all():
+    """Refresh JWTs for all active accounts."""
+    accounts = await db.get_active_accounts()
+    results = {"refreshed": 0, "failed": 0, "total": len(accounts)}
+    for acct in accounts:
+        new_jwt = await pool.refresh_jwt(acct)
+        if new_jwt:
+            results["refreshed"] += 1
+        else:
+            results["failed"] += 1
+    return results
 
 
 # ---------------------------------------------------------------------------

emergent2api/backends/__init__.py

@@ -0,0 +1 @@
+

emergent2api/backends/jobs.py

@@ -1,4 +1,10 @@
-"""Jobs API backend: submit task + poll trajectory for responses."""
+"""Jobs API backend: submit task + poll trajectory for responses.
+
+Streaming uses the "longest text" approach: each poll retrieves trajectory
+items and picks the longest text across multiple possible fields. Delta text
+(new chars since last poll) is yielded to the caller.  End-of-stream is
+detected via a double-confirmation mechanism (stable content for N cycles).
+"""
 from __future__ import annotations
 
 import asyncio
@@ -15,6 +21,19 @@ from .base import EmergentBackend
 
 logger = logging.getLogger("emergent2api.jobs")
 
+_TEXT_KEYS = ("thought", "output", "message", "content", "text", "response")
+
+
+def _extract_text_from_item(item: dict) -> str:
+    """Try multiple fields to extract the longest text from a trajectory item."""
+    payload = item.get("traj_payload") or item
+    best = ""
+    for key in _TEXT_KEYS:
+        val = payload.get(key)
+        if val and isinstance(val, str) and len(val) > len(best):
+            best = val
+    return best
+
 
 class JobsBackend(EmergentBackend):
     """Wraps Emergent's Jobs API (submit-queue + trajectory polling) into a chat interface."""
@@ -28,11 +47,8 @@ class JobsBackend(EmergentBackend):
         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
         )
@@ -42,66 +58,69 @@ class JobsBackend(EmergentBackend):
 
         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
+        consecutive_unchanged = 0
+        confirmed_end = False
+        has_received_anything = False
+        max_wait_cycles = 400
 
-        while (time.time() - start_time) < settings.poll_timeout:
-            await asyncio.sleep(settings.poll_interval)
+        # Initial wait to let the container spin up
+        await asyncio.sleep(8.0)
 
+        for _ in range(max_wait_cycles):
             result = await asyncio.to_thread(self._poll_trajectory, jwt, ref_id)
             if result is None:
-                empty_polls += 1
-                if empty_polls > 5:
-                    continue
+                await self._dynamic_sleep(consecutive_unchanged)
+                consecutive_unchanged += 1
                 continue
 
             data_items = result.get("data", [])
-            job_status = result.get("status", "")
 
-            new_content = False
+            # Find the longest text across all items (main content)
+            current_full_text = ""
+            current_thinking = ""
             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", {})
+                text = _extract_text_from_item(item)
+                if len(text) > len(current_full_text):
+                    current_full_text = text
+                payload = item.get("traj_payload") or {}
                 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"):
+                if reasoning and len(reasoning) > len(current_thinking):
+                    current_thinking = reasoning
+
+            # Yield thinking delta
+            if current_thinking and len(current_thinking) > len(full_thinking):
+                delta = current_thinking[len(full_thinking):]
+                full_thinking = current_thinking
+                if stream and delta:
+                    yield {"type": "thinking", "content": delta}
+
+            # Yield text delta
+            if len(current_full_text) > len(full_text):
+                delta = current_full_text[len(full_text):]
+                full_text = current_full_text
+                consecutive_unchanged = 0
+                confirmed_end = False
+                has_received_anything = True
+                if stream and delta:
+                    yield {"type": "text", "content": delta}
+            else:
+                consecutive_unchanged += 1
+
+                if has_received_anything and full_text and consecutive_unchanged >= 6:
+                    if not confirmed_end:
+                        logger.debug("Content stable, confirming end...")
+                        confirmed_end = True
+                        await asyncio.sleep(5.0)
+                        continue
+                    logger.info("Stream completed (double-confirmed stable)")
                     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
+
+            await self._dynamic_sleep(consecutive_unchanged)
+
+        if not has_received_anything:
+            logger.warning("Stream finished with NO content")
 
         yield {
             "type": "done",
@@ -109,6 +128,15 @@ class JobsBackend(EmergentBackend):
             "thinking": full_thinking,
         }
 
+    @staticmethod
+    async def _dynamic_sleep(consecutive_unchanged: int) -> None:
+        if consecutive_unchanged == 0:
+            await asyncio.sleep(1.0)
+        elif consecutive_unchanged < 3:
+            await asyncio.sleep(2.0)
+        else:
+            await asyncio.sleep(3.0)
+
     def _submit_job(
         self,
         jwt: str,

emergent2api/config.py

@@ -7,7 +7,7 @@ from dataclasses import dataclass, field
 
 @dataclass
 class Settings:
-    api_key: str = field(default_factory=lambda: os.environ.get("EMERGENT2API_KEY", "sk-emergent2api"))
+    api_key: str = field(default_factory=lambda: os.environ.get("EMERGENT2API_KEY", "sk-6loA0HwMQP1mdhPvI"))
     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")))
@@ -28,7 +28,7 @@ class Settings:
     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"
+    app_version: str = "3.5.0"
 
 
 settings = Settings()

emergent2api/database.py

@@ -118,6 +118,37 @@ async def update_jwt(account_id: int, jwt: str, refresh_token: str = "") -> None
         )
 
 
+async def get_all_accounts() -> list[dict[str, Any]]:
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        rows = await conn.fetch("SELECT * FROM emergent_accounts ORDER BY id ASC")
+        return [dict(r) for r in rows]
+
+
+async def get_account_by_id(account_id: int) -> Optional[dict[str, Any]]:
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow("SELECT * FROM emergent_accounts WHERE id = $1", account_id)
+        return dict(row) if row else None
+
+
+async def delete_account(account_id: int) -> None:
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        await conn.execute("DELETE FROM emergent_accounts WHERE id = $1", account_id)
+
+
+async def toggle_account(account_id: int) -> bool:
+    """Toggle active status and return the new state."""
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(
+            "UPDATE emergent_accounts SET is_active = NOT is_active, updated_at = NOW() WHERE id = $1 RETURNING is_active",
+            account_id,
+        )
+        return row["is_active"] if row else False
+
+
 async def get_account_count() -> dict[str, int]:
     pool = await get_pool()
     async with pool.acquire() as conn:

emergent2api/routes/__init__.py

@@ -0,0 +1 @@
+

requirements.txt

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