Initial: Agentic backend with NIM + tools

Dockerfile

@@ -0,0 +1,28 @@
+FROM python:3.11-slim
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    git bash curl && \
+    rm -rf /var/lib/apt/lists/*
+
+# Create non-root user (HF requirement: uid 1000)
+RUN useradd -m -u 1000 user
+USER user
+ENV HOME=/home/user \
+    PATH=/home/user/.local/bin:$PATH
+
+WORKDIR $HOME/app
+
+# Install Python dependencies
+COPY --chown=user requirements.txt .
+RUN pip install --no-cache-dir --user -r requirements.txt
+
+# Copy application
+COPY --chown=user backend.py .
+
+# Create workspace directory
+RUN mkdir -p /tmp/workspace
+
+EXPOSE 7860
+
+CMD ["python", "backend.py"]

README.md

@@ -0,0 +1,14 @@
+---
+title: claude-code-backend
+emoji: 🤖
+colorFrom: blue
+colorTo: purple
+sdk: docker
+app_port: 7860
+---
+
+# Claude Code Backend
+
+Agentic coding backend powered by NVIDIA NIM models. Provides Claude-Code-like capabilities (Read, Write, Edit, Bash, Grep) through an OpenAI-compatible API.
+
+Used as a backend provider for [better-chatbot](https://huggingface.co/spaces/augment17/better-chatbot).

backend.py

@@ -0,0 +1,668 @@
+"""
+Claude Code Backend — Agentic coding backend powered by NVIDIA NIM models.
+Exposes an OpenAI-compatible /v1/chat/completions endpoint with built-in
+tools for file operations and bash execution.
+
+Architecture:
+  Space 1 (better-chatbot) --> this backend --> NVIDIA NIM API
+  
+The agentic loop:
+  1. Receive user message from Space 1
+  2. Send to NIM model with tool definitions
+  3. If model returns tool_calls, execute them and loop
+  4. If model returns text, stream it back to Space 1
+  5. Persist conversation in Postgres
+"""
+
+import os
+import json
+import uuid
+import subprocess
+import asyncio
+import time
+from pathlib import Path
+from typing import AsyncIterator, Optional
+
+from fastapi import FastAPI, Request, Header, HTTPException
+from fastapi.responses import StreamingResponse, JSONResponse
+from fastapi.middleware.cors import CORSMiddleware
+from openai import AsyncOpenAI
+import asyncpg
+
+# ---------------------------------------------------------------------------
+# Configuration
+# ---------------------------------------------------------------------------
+
+NIM_API_KEY = os.environ.get("NVIDIA_NIM_API_KEY", "")
+BACKEND_API_KEY = os.environ.get("BACKEND_API_KEY", "")
+DATABASE_URL = os.environ.get("DATABASE_URL", "")
+WORKSPACE_DIR = os.environ.get("WORKSPACE_DIR", "/tmp/workspace")
+MAX_TOOL_ROUNDS = int(os.environ.get("MAX_TOOL_ROUNDS", "10"))
+
+# NIM models that reliably support tool/function calling
+TOOL_CAPABLE_MODELS = {
+    "meta/llama-3.1-70b-instruct": "Llama 3.1 70B (Agentic)",
+    "meta/llama-3.1-405b-instruct": "Llama 3.1 405B (Agentic)",
+    "qwen/qwen2.5-coder-32b-instruct": "Qwen 2.5 Coder 32B (Agentic)",
+    "nvidia/llama-3.1-nemotron-70b-instruct": "Nemotron 70B (Agentic)",
+    "meta/llama-3.3-70b-instruct": "Llama 3.3 70B (Agentic)",
+}
+
+# All models (tool-capable get agentic mode, others get plain chat)
+ALL_MODELS = {
+    **TOOL_CAPABLE_MODELS,
+    "deepseek-ai/deepseek-r1": "DeepSeek R1 (Chat only)",
+    "mistralai/mistral-large-2-instruct": "Mistral Large 2 (Chat only)",
+}
+
+# Ensure workspace exists
+Path(WORKSPACE_DIR).mkdir(parents=True, exist_ok=True)
+
+# ---------------------------------------------------------------------------
+# NIM Client
+# ---------------------------------------------------------------------------
+
+nim_client = AsyncOpenAI(
+    base_url="https://integrate.api.nvidia.com/v1",
+    api_key=NIM_API_KEY,
+)
+
+# ---------------------------------------------------------------------------
+# Tool Definitions (OpenAI function calling format)
+# ---------------------------------------------------------------------------
+
+TOOLS = [
+    {
+        "type": "function",
+        "function": {
+            "name": "read_file",
+            "description": "Read the contents of a file. Use this to inspect existing code, configs, or any text file.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {
+                        "type": "string",
+                        "description": "Relative path to the file from the workspace root"
+                    }
+                },
+                "required": ["path"]
+            }
+        }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "write_file",
+            "description": "Write content to a file. Creates the file if it doesn't exist, overwrites if it does. Creates parent directories automatically.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {
+                        "type": "string",
+                        "description": "Relative path to the file from the workspace root"
+                    },
+                    "content": {
+                        "type": "string",
+                        "description": "The full content to write to the file"
+                    }
+                },
+                "required": ["path", "content"]
+            }
+        }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "run_bash",
+            "description": "Execute a bash command in the workspace directory. Use for installing packages, running scripts, git operations, etc. Commands run with a 30 second timeout.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "command": {
+                        "type": "string",
+                        "description": "The bash command to execute"
+                    }
+                },
+                "required": ["command"]
+            }
+        }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "list_directory",
+            "description": "List files and directories in a given path. Shows file sizes and directory markers.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {
+                        "type": "string",
+                        "description": "Relative path to the directory from workspace root. Use '.' for the workspace root."
+                    }
+                },
+                "required": ["path"]
+            }
+        }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "grep_search",
+            "description": "Search for a pattern in files within the workspace. Returns matching lines with file paths and line numbers.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "pattern": {
+                        "type": "string",
+                        "description": "The search pattern (supports basic regex)"
+                    },
+                    "path": {
+                        "type": "string",
+                        "description": "Directory or file to search in, relative to workspace root. Defaults to '.'",
+                    }
+                },
+                "required": ["pattern"]
+            }
+        }
+    },
+]
+
+# ---------------------------------------------------------------------------
+# Tool Execution
+# ---------------------------------------------------------------------------
+
+def _safe_path(rel_path: str) -> Path:
+    """Resolve a relative path safely within the workspace."""
+    workspace = Path(WORKSPACE_DIR).resolve()
+    target = (workspace / rel_path).resolve()
+    # Prevent path traversal
+    if not str(target).startswith(str(workspace)):
+        raise ValueError(f"Path traversal detected: {rel_path}")
+    return target
+
+
+def execute_tool(name: str, arguments: dict) -> str:
+    """Execute a tool and return its output as a string."""
+    try:
+        if name == "read_file":
+            path = _safe_path(arguments["path"])
+            if not path.exists():
+                return f"Error: File not found: {arguments['path']}"
+            if not path.is_file():
+                return f"Error: Not a file: {arguments['path']}"
+            content = path.read_text(encoding="utf-8", errors="replace")
+            if len(content) > 50000:
+                return content[:50000] + f"\n\n[Truncated — file is {len(content)} chars]"
+            return content
+
+        elif name == "write_file":
+            path = _safe_path(arguments["path"])
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(arguments["content"], encoding="utf-8")
+            return f"Successfully wrote {len(arguments['content'])} chars to {arguments['path']}"
+
+        elif name == "run_bash":
+            command = arguments["command"]
+            # Safety: block dangerous commands
+            blocked = ["rm -rf /", "mkfs", "dd if=", ":(){", "fork bomb"]
+            if any(b in command.lower() for b in blocked):
+                return "Error: Command blocked for safety reasons"
+            result = subprocess.run(
+                ["bash", "-c", command],
+                cwd=WORKSPACE_DIR,
+                capture_output=True,
+                text=True,
+                timeout=30,
+                env={**os.environ, "HOME": "/tmp", "PATH": os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin")},
+            )
+            output = ""
+            if result.stdout:
+                output += result.stdout
+            if result.stderr:
+                output += ("\n" if output else "") + f"[stderr] {result.stderr}"
+            if result.returncode != 0:
+                output += f"\n[exit code: {result.returncode}]"
+            if not output:
+                output = "[command completed with no output]"
+            # Truncate very long outputs
+            if len(output) > 20000:
+                output = output[:20000] + f"\n\n[Truncated — output is {len(output)} chars]"
+            return output
+
+        elif name == "list_directory":
+            path = _safe_path(arguments.get("path", "."))
+            if not path.exists():
+                return f"Error: Directory not found: {arguments.get('path', '.')}"
+            if not path.is_dir():
+                return f"Error: Not a directory: {arguments.get('path', '.')}"
+            entries = []
+            for item in sorted(path.iterdir()):
+                if item.is_dir():
+                    entries.append(f"  📁 {item.name}/")
+                else:
+                    size = item.stat().st_size
+                    if size < 1024:
+                        size_str = f"{size}B"
+                    elif size < 1024 * 1024:
+                        size_str = f"{size/1024:.1f}KB"
+                    else:
+                        size_str = f"{size/(1024*1024):.1f}MB"
+                    entries.append(f"  📄 {item.name} ({size_str})")
+            return f"Contents of {arguments.get('path', '.')}:\n" + "\n".join(entries) if entries else "Empty directory"
+
+        elif name == "grep_search":
+            pattern = arguments["pattern"]
+            search_path = arguments.get("path", ".")
+            path = _safe_path(search_path)
+            result = subprocess.run(
+                ["grep", "-rn", "--include=*", pattern, str(path)],
+                capture_output=True,
+                text=True,
+                timeout=10,
+                cwd=WORKSPACE_DIR,
+            )
+            output = result.stdout if result.stdout else "No matches found"
+            if len(output) > 10000:
+                output = output[:10000] + "\n\n[Truncated]"
+            return output
+
+        else:
+            return f"Error: Unknown tool: {name}"
+
+    except subprocess.TimeoutExpired:
+        return "Error: Command timed out after 30 seconds"
+    except ValueError as e:
+        return f"Error: {str(e)}"
+    except Exception as e:
+        return f"Error executing {name}: {str(e)}"
+
+
+# ---------------------------------------------------------------------------
+# Database (Session Persistence)
+# ---------------------------------------------------------------------------
+
+db_pool: Optional[asyncpg.Pool] = None
+
+
+async def init_db():
+    """Initialize database connection pool and create tables."""
+    global db_pool
+    if not DATABASE_URL:
+        return
+    try:
+        db_pool = await asyncpg.create_pool(DATABASE_URL, ssl="require", min_size=1, max_size=5)
+        async with db_pool.acquire() as conn:
+            await conn.execute("""
+                CREATE TABLE IF NOT EXISTS agent_sessions (
+                    id BIGSERIAL PRIMARY KEY,
+                    session_id TEXT NOT NULL,
+                    role TEXT NOT NULL,
+                    content TEXT,
+                    tool_calls JSONB,
+                    tool_call_id TEXT,
+                    created_at TIMESTAMPTZ DEFAULT NOW()
+                );
+                CREATE INDEX IF NOT EXISTS idx_agent_sessions_sid ON agent_sessions(session_id);
+            """)
+    except Exception as e:
+        print(f"[DB] Warning: Could not initialize database: {e}")
+        db_pool = None
+
+
+async def save_message(session_id: str, role: str, content: str = None, 
+                       tool_calls: list = None, tool_call_id: str = None):
+    """Save a message to the session store."""
+    if not db_pool:
+        return
+    try:
+        async with db_pool.acquire() as conn:
+            await conn.execute(
+                "INSERT INTO agent_sessions (session_id, role, content, tool_calls, tool_call_id) VALUES ($1, $2, $3, $4, $5)",
+                session_id, role, content,
+                json.dumps(tool_calls) if tool_calls else None,
+                tool_call_id,
+            )
+    except Exception as e:
+        print(f"[DB] Warning: Could not save message: {e}")
+
+
+async def load_session(session_id: str) -> list:
+    """Load conversation history from the session store."""
+    if not db_pool:
+        return []
+    try:
+        async with db_pool.acquire() as conn:
+            rows = await conn.fetch(
+                "SELECT role, content, tool_calls, tool_call_id FROM agent_sessions WHERE session_id = $1 ORDER BY id",
+                session_id,
+            )
+            messages = []
+            for row in rows:
+                msg = {"role": row["role"]}
+                if row["content"]:
+                    msg["content"] = row["content"]
+                if row["tool_calls"]:
+                    msg["tool_calls"] = json.loads(row["tool_calls"])
+                if row["tool_call_id"]:
+                    msg["tool_call_id"] = row["tool_call_id"]
+                messages.append(msg)
+            return messages
+    except Exception as e:
+        print(f"[DB] Warning: Could not load session: {e}")
+        return []
+
+
+# ---------------------------------------------------------------------------
+# SSE Chunk Formatting (OpenAI delta format)
+# ---------------------------------------------------------------------------
+
+def make_chunk(request_id: str, model: str, content: str = "", finish_reason: str = None) -> str:
+    """Create an OpenAI-compatible SSE chunk."""
+    delta = {}
+    if content:
+        delta["content"] = content
+    if finish_reason and not content:
+        delta = {}
+    
+    chunk = {
+        "id": f"chatcmpl-{request_id}",
+        "object": "chat.completion.chunk",
+        "created": int(time.time()),
+        "model": model,
+        "choices": [{
+            "index": 0,
+            "delta": delta,
+            "finish_reason": finish_reason,
+        }],
+    }
+    return f"data: {json.dumps(chunk)}\n\n"
+
+
+# ---------------------------------------------------------------------------
+# FastAPI Application
+# ---------------------------------------------------------------------------
+
+app = FastAPI(title="Claude Code Backend", version="1.0.0")
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+def auth(authorization: str = None):
+    """Verify bearer token."""
+    if not BACKEND_API_KEY:
+        return  # No auth configured
+    expected = f"Bearer {BACKEND_API_KEY}"
+    if authorization != expected:
+        raise HTTPException(status_code=401, detail="Unauthorized")
+
+
+@app.on_event("startup")
+async def startup():
+    await init_db()
+    Path(WORKSPACE_DIR).mkdir(parents=True, exist_ok=True)
+    print(f"[Backend] Started. Workspace: {WORKSPACE_DIR}")
+    print(f"[Backend] Tool-capable models: {list(TOOL_CAPABLE_MODELS.keys())}")
+    print(f"[Backend] DB connected: {db_pool is not None}")
+
+
+# ---------------------------------------------------------------------------
+# /v1/chat/completions — Main endpoint
+# ---------------------------------------------------------------------------
+
+AGENTIC_SYSTEM_PROMPT = """You are an expert coding assistant with access to tools for file operations and command execution. 
+
+When the user asks you to create, edit, or debug code:
+1. Use `list_directory` and `read_file` to understand the current state
+2. Use `write_file` to create or modify files
+3. Use `run_bash` to execute commands (install packages, run scripts, test code)
+4. Use `grep_search` to find patterns in code
+
+IMPORTANT RULES:
+- Always use tools to take action. Do NOT just describe what to do — actually DO it.
+- After writing code, run it to verify it works.
+- If a command fails, read the error and fix it.
+- Work in the /tmp/workspace directory.
+- Be concise in your explanations, but thorough in your tool usage.
+"""
+
+
+@app.post("/v1/chat/completions")
+async def chat_completions(request: Request, authorization: str = Header(None)):
+    auth(authorization)
+    body = await request.json()
+
+    requested_model = body.get("model", "meta/llama-3.1-70b-instruct")
+    messages = body.get("messages", [])
+    stream = body.get("stream", False)
+    session_id = body.get("session_id") or str(uuid.uuid4())
+    
+    is_agentic = requested_model in TOOL_CAPABLE_MODELS
+    request_id = str(uuid.uuid4())[:8]
+
+    # Build message history
+    final_messages = []
+    
+    # Add agentic system prompt for tool-capable models
+    if is_agentic:
+        # Check if there's already a system message
+        has_system = any(m.get("role") == "system" for m in messages)
+        if has_system:
+            # Prepend agentic prompt to existing system message
+            for m in messages:
+                if m["role"] == "system":
+                    final_messages.append({
+                        "role": "system",
+                        "content": AGENTIC_SYSTEM_PROMPT + "\n\nAdditional instructions:\n" + m["content"]
+                    })
+                else:
+                    final_messages.append(m)
+        else:
+            final_messages.append({"role": "system", "content": AGENTIC_SYSTEM_PROMPT})
+            final_messages.extend(messages)
+    else:
+        final_messages = list(messages)
+
+    # Save the user's message to DB
+    user_msg = next((m for m in reversed(messages) if m.get("role") == "user"), None)
+    if user_msg:
+        await save_message(session_id, "user", user_msg.get("content", ""))
+
+    if not stream:
+        # Non-streaming: simple completion
+        try:
+            kwargs = {"model": requested_model, "messages": final_messages}
+            if is_agentic:
+                kwargs["tools"] = TOOLS
+                kwargs["tool_choice"] = "auto"
+            response = await nim_client.chat.completions.create(**kwargs)
+            content = response.choices[0].message.content or ""
+            await save_message(session_id, "assistant", content)
+            return JSONResponse({
+                "id": f"chatcmpl-{request_id}",
+                "object": "chat.completion",
+                "created": int(time.time()),
+                "model": requested_model,
+                "choices": [{"index": 0, "message": {"role": "assistant", "content": content}, "finish_reason": "stop"}],
+            })
+        except Exception as e:
+            return JSONResponse({"error": {"message": str(e), "type": "internal_error"}}, status_code=500)
+
+    # Streaming + agentic loop
+    async def generate() -> AsyncIterator[str]:
+        nonlocal final_messages
+        try:
+            for round_num in range(MAX_TOOL_ROUNDS + 1):
+                kwargs = {"model": requested_model, "messages": final_messages, "stream": True}
+                if is_agentic:
+                    kwargs["tools"] = TOOLS
+                    kwargs["tool_choice"] = "auto"
+
+                # Collect streamed response
+                full_content = ""
+                tool_calls_raw = {}  # index -> {id, name, arguments_str}
+                
+                async for chunk in await nim_client.chat.completions.create(**kwargs):
+                    choice = chunk.choices[0] if chunk.choices else None
+                    if not choice:
+                        continue
+                    delta = choice.delta
+                    
+                    # Stream text content to client
+                    if delta and delta.content:
+                        full_content += delta.content
+                        yield make_chunk(request_id, requested_model, delta.content)
+                    
+                    # Collect tool calls
+                    if delta and delta.tool_calls:
+                        for tc in delta.tool_calls:
+                            idx = tc.index
+                            if idx not in tool_calls_raw:
+                                tool_calls_raw[idx] = {
+                                    "id": tc.id or f"call_{uuid.uuid4().hex[:8]}",
+                                    "name": tc.function.name if tc.function and tc.function.name else "",
+                                    "arguments": ""
+                                }
+                            if tc.function and tc.function.name:
+                                tool_calls_raw[idx]["name"] = tc.function.name
+                            if tc.id:
+                                tool_calls_raw[idx]["id"] = tc.id
+                            if tc.function and tc.function.arguments:
+                                tool_calls_raw[idx]["arguments"] += tc.function.arguments
+                    
+                    # Check for finish
+                    if choice.finish_reason == "stop":
+                        break
+                    if choice.finish_reason == "tool_calls":
+                        break
+
+                # If no tool calls, we're done
+                if not tool_calls_raw:
+                    await save_message(session_id, "assistant", full_content)
+                    yield make_chunk(request_id, requested_model, finish_reason="stop")
+                    yield "data: [DONE]\n\n"
+                    return
+
+                # Execute tool calls
+                tool_calls_list = []
+                for idx in sorted(tool_calls_raw.keys()):
+                    tc = tool_calls_raw[idx]
+                    tool_calls_list.append({
+                        "id": tc["id"],
+                        "type": "function",
+                        "function": {"name": tc["name"], "arguments": tc["arguments"]}
+                    })
+
+                # Add assistant message with tool calls to history
+                assistant_msg = {"role": "assistant", "content": full_content or None, "tool_calls": tool_calls_list}
+                final_messages.append(assistant_msg)
+
+                # Execute each tool and add results
+                for tc in tool_calls_list:
+                    func_name = tc["function"]["name"]
+                    try:
+                        func_args = json.loads(tc["function"]["arguments"])
+                    except json.JSONDecodeError:
+                        func_args = {}
+                    
+                    # Show tool execution to user
+                    yield make_chunk(request_id, requested_model, f"\n\n🔧 **{func_name}**")
+                    if func_name == "run_bash" and "command" in func_args:
+                        yield make_chunk(request_id, requested_model, f": `{func_args['command']}`\n")
+                    elif func_name == "read_file" and "path" in func_args:
+                        yield make_chunk(request_id, requested_model, f": `{func_args['path']}`\n")
+                    elif func_name == "write_file" and "path" in func_args:
+                        yield make_chunk(request_id, requested_model, f": `{func_args['path']}`\n")
+                    elif func_name == "list_directory":
+                        yield make_chunk(request_id, requested_model, f": `{func_args.get('path', '.')}`\n")
+                    elif func_name == "grep_search":
+                        yield make_chunk(request_id, requested_model, f": `{func_args.get('pattern', '')}`\n")
+                    else:
+                        yield make_chunk(request_id, requested_model, "\n")
+
+                    # Execute the tool
+                    result = execute_tool(func_name, func_args)
+
+                    # Show truncated result to user
+                    preview = result[:500] + ("..." if len(result) > 500 else "")
+                    yield make_chunk(request_id, requested_model, f"```\n{preview}\n```\n")
+
+                    # Add tool result to message history
+                    final_messages.append({
+                        "role": "tool",
+                        "tool_call_id": tc["id"],
+                        "content": result,
+                    })
+
+                    await save_message(session_id, "tool", result, tool_call_id=tc["id"])
+
+                # Continue the agentic loop (model processes tool results)
+
+            # If we hit max rounds, finish
+            yield make_chunk(request_id, requested_model, "\n\n⚠️ Reached maximum tool call rounds.")
+            yield make_chunk(request_id, requested_model, finish_reason="stop")
+            yield "data: [DONE]\n\n"
+
+        except Exception as e:
+            error_msg = f"\n\n❌ Error: {str(e)}"
+            yield make_chunk(request_id, requested_model, error_msg)
+            yield make_chunk(request_id, requested_model, finish_reason="stop")
+            yield "data: [DONE]\n\n"
+
+    return StreamingResponse(
+        generate(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",
+            "Connection": "keep-alive",
+        },
+    )
+
+
+# ---------------------------------------------------------------------------
+# /v1/models — Model listing
+# ---------------------------------------------------------------------------
+
+@app.get("/v1/models")
+async def list_models(authorization: str = Header(None)):
+    auth(authorization)
+    models = []
+    for model_id, display_name in ALL_MODELS.items():
+        models.append({
+            "id": model_id,
+            "object": "model",
+            "created": 1700000000,
+            "owned_by": "nvidia-nim",
+            "permission": [],
+            "root": model_id,
+            "parent": None,
+        })
+    return {"object": "list", "data": models}
+
+
+# ---------------------------------------------------------------------------
+# /health — Health check
+# ---------------------------------------------------------------------------
+
+@app.get("/health")
+async def health():
+    return {
+        "status": "ok",
+        "workspace": WORKSPACE_DIR,
+        "workspace_exists": Path(WORKSPACE_DIR).exists(),
+        "db_connected": db_pool is not None,
+        "models_count": len(ALL_MODELS),
+    }
+
+
+# ---------------------------------------------------------------------------
+# Entrypoint
+# ---------------------------------------------------------------------------
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860)

requirements.txt

@@ -0,0 +1,5 @@
+fastapi==0.115.12
+uvicorn[standard]==0.34.2
+openai==1.86.0
+asyncpg==0.30.0
+anyio==4.9.0