split api and app

api.py

@@ -0,0 +1,134 @@
+import os
+from datetime import datetime
+from pathlib import Path
+
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+
+from avatar_store import avatar_generated_path
+from generate_image_with_nano import build_prompt, run_edit
+from mcp_tools import (
+    create_avatar,
+    delete_avatar_memory,
+    delete_avatar_portrait,
+    delete_generated_images,
+    ensure_public_avatar,
+    generate_as_avatar,
+    get_avatar,
+    get_avatar_context,
+    record_generated_image,
+    retrieve_avatar_snippets,
+    set_avatar_portrait,
+    store_avatar_memory,
+    summarize_avatar_context,
+)
+
+fastapi_app = FastAPI()
+fastapi_app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+def _handle(func, payload):
+    try:
+        return func(payload or {})
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc))
+
+
+def _resolve_path(path_str):
+    if not path_str:
+        return None
+    path = Path(path_str)
+    if not path.is_absolute():
+        path = Path(__file__).parent / path
+    return str(path)
+
+
+@fastapi_app.post("/mcp/create_avatar")
+def api_create_avatar(payload: dict):
+    return _handle(create_avatar, payload)
+
+
+@fastapi_app.post("/mcp/get_avatar")
+def api_get_avatar(payload: dict):
+    return _handle(get_avatar, payload)
+
+
+@fastapi_app.post("/mcp/generate_as_avatar")
+def api_generate_as_avatar(payload: dict):
+    return _handle(generate_as_avatar, payload)
+
+
+@fastapi_app.post("/mcp/generate_image")
+def api_generate_image(payload: dict):
+    payload = payload or {}
+    avatar_id = payload.get("avatar_id")
+    if not avatar_id:
+        raise HTTPException(status_code=400, detail="avatar_id required")
+    message = (payload.get("message") or "").strip()
+    reply = (payload.get("reply") or "").strip()
+    context = message or reply or "scene with the avatar"
+    try:
+        avatar = ensure_public_avatar(avatar_id)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc))
+    portrait_path = _resolve_path(avatar.get("portrait"))
+    if not portrait_path or not Path(portrait_path).exists():
+        raise HTTPException(status_code=400, detail="portrait not found for avatar")
+    prompt = build_prompt(avatar.get("persona", "Avatar"), context)
+    timestamp = datetime.utcnow().strftime("%Y%m%d-%H%M%S")
+    out_path = avatar_generated_path(avatar_id, timestamp)
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    try:
+        rc = run_edit(Path(portrait_path), prompt, out_path)
+    except SystemExit:
+        rc = 1
+    if rc != 0:
+        raise HTTPException(status_code=500, detail="image generation failed")
+    record_generated_image(avatar_id, out_path, prompt, timestamp)
+    return {
+        "status": "generated",
+        "path": str(out_path),
+        "prompt": prompt,
+        "timestamp": timestamp,
+    }
+
+
+@fastapi_app.post("/mcp/store_avatar_memory")
+def api_store_avatar_memory(payload: dict):
+    return _handle(store_avatar_memory, payload)
+
+
+@fastapi_app.post("/mcp/delete_avatar_memory")
+def api_delete_avatar_memory(payload: dict):
+    return _handle(delete_avatar_memory, payload)
+
+
+@fastapi_app.post("/mcp/get_avatar_context")
+def api_get_avatar_context(payload: dict):
+    return _handle(get_avatar_context, payload)
+
+
+@fastapi_app.post("/mcp/delete_generated_images")
+def api_delete_generated_images(payload: dict):
+    return _handle(delete_generated_images, payload)
+
+
+@fastapi_app.post("/mcp/retrieve_snippets")
+def api_retrieve_snippets(payload: dict):
+    return _handle(retrieve_avatar_snippets, payload)
+
+
+@fastapi_app.post("/mcp/summarize_avatar")
+def api_summarize_avatar(payload: dict):
+    return _handle(summarize_avatar_context, payload)
+
+
+@fastapi_app.get("/")
+def root():
+    return {"status": "ok"}

app.py

@@ -4,44 +4,23 @@ from datetime import datetime
 from pathlib import Path
 
 import gradio as gr
-from fastapi import FastAPI, HTTPException
-from fastapi.middleware.cors import CORSMiddleware
 
+from api import fastapi_app
 from avatar_store import avatar_generated_path
+from generate_image_with_nano import build_prompt, run_edit
 from mcp_tools import (
     create_avatar,
     delete_avatar_memory,
     delete_avatar_portrait,
-    ensure_public_avatar,
+    delete_generated_images,
     generate_as_avatar,
     get_avatar,
-    get_avatar_context,
     record_generated_image,
-    delete_generated_images,
     set_avatar_portrait,
     store_avatar_memory,
-    retrieve_avatar_snippets,
-    summarize_avatar_context,
-)
-from generate_image_with_nano import build_prompt, run_edit
-
-fastapi_app = FastAPI()
-fastapi_app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
 )
 
 
-def _handle(func, payload):
-    try:
-        return func(payload or {})
-    except ValueError as exc:
-        raise HTTPException(status_code=400, detail=str(exc))
-
-
 def _resolve_path(path_str):
     if not path_str:
         return None
@@ -51,86 +30,6 @@ def _resolve_path(path_str):
     return str(path)
 
 
-@fastapi_app.post("/mcp/create_avatar")
-def api_create_avatar(payload: dict):
-    return _handle(create_avatar, payload)
-
-
-@fastapi_app.post("/mcp/get_avatar")
-def api_get_avatar(payload: dict):
-    return _handle(get_avatar, payload)
-
-
-@fastapi_app.post("/mcp/generate_as_avatar")
-def api_generate_as_avatar(payload: dict):
-    return _handle(generate_as_avatar, payload)
-
-
-@fastapi_app.post("/mcp/generate_image")
-def api_generate_image(payload: dict):
-    payload = payload or {}
-    avatar_id = payload.get("avatar_id")
-    if not avatar_id:
-        raise HTTPException(status_code=400, detail="avatar_id required")
-    message = (payload.get("message") or "").strip()
-    reply = (payload.get("reply") or "").strip()
-    context = message or reply or "scene with the avatar"
-    try:
-        avatar = ensure_public_avatar(avatar_id)
-    except ValueError as exc:
-        raise HTTPException(status_code=400, detail=str(exc))
-    portrait_path = _resolve_path(avatar.get("portrait"))
-    if not portrait_path or not Path(portrait_path).exists():
-        raise HTTPException(status_code=400, detail="portrait not found for avatar")
-    prompt = build_prompt(avatar.get("persona", "Avatar"), context)
-    timestamp = datetime.utcnow().strftime("%Y%m%d-%H%M%S")
-    out_path = avatar_generated_path(avatar_id, timestamp)
-    out_path.parent.mkdir(parents=True, exist_ok=True)
-    try:
-        rc = run_edit(Path(portrait_path), prompt, out_path)
-    except SystemExit:
-        rc = 1
-    if rc != 0:
-        raise HTTPException(status_code=500, detail="image generation failed")
-    record_generated_image(avatar_id, out_path, prompt, timestamp)
-    return {
-        "status": "generated",
-        "path": str(out_path),
-        "prompt": prompt,
-        "timestamp": timestamp,
-    }
-
-
-@fastapi_app.post("/mcp/store_avatar_memory")
-def api_store_avatar_memory(payload: dict):
-    return _handle(store_avatar_memory, payload)
-
-
-@fastapi_app.post("/mcp/delete_avatar_memory")
-def api_delete_avatar_memory(payload: dict):
-    return _handle(delete_avatar_memory, payload)
-
-
-@fastapi_app.post("/mcp/get_avatar_context")
-def api_get_avatar_context(payload: dict):
-    return _handle(get_avatar_context, payload)
-
-
-@fastapi_app.post("/mcp/delete_generated_images")
-def api_delete_generated_images(payload: dict):
-    return _handle(delete_generated_images, payload)
-
-
-@fastapi_app.post("/mcp/retrieve_snippets")
-def api_retrieve_snippets(payload: dict):
-    return _handle(retrieve_avatar_snippets, payload)
-
-
-@fastapi_app.post("/mcp/summarize_avatar")
-def api_summarize_avatar(payload: dict):
-    return _handle(summarize_avatar_context, payload)
-
-
 def decide_tool(message: str) -> str:
     text = (message or "").lower()
     if "remember" in text:
@@ -535,11 +434,6 @@ with gr.Blocks() as ui:
 app = gr.mount_gradio_app(fastapi_app, ui, path="/")
 
 
-@fastapi_app.get("/")
-def root():
-    return {"status": "ok"}
-
-
 if __name__ == "__main__" and not os.getenv("SPACE_ID"):
     import uvicorn
 

requirements.txt

@@ -4,3 +4,4 @@ uvicorn
 httpx
 google-genai
 Pillow
+pydantic>=2,<3