Upload folder using huggingface_hub

.hfignore

@@ -0,0 +1,9 @@
+.git/
+.github/
+docs/
+sql/
+test_gemini_tools.py
+.env.example
+.gitignore
+ROADMAP.md
+pyproject.toml

Dockerfile

@@ -0,0 +1,21 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+ENV PYTHONPATH=/app
+ENV OPENCODE_STORAGE_PATH=/app
+
+RUN chmod -R 777 /app
+
+EXPOSE 7860
+
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,10 +1,100 @@
 ---
-title: Opencode Api
-emoji: 📊
-colorFrom: yellow
-colorTo: gray
+title: opencode-api
+emoji: 🤖
+colorFrom: blue
+colorTo: purple
 sdk: docker
+app_port: 7860
 pinned: false
+license: mit
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# OpenCode API
+
+LLM Agent API Server - ported from TypeScript [opencode](https://github.com/anomalyco/opencode) to Python.
+
+## Features
+
+- **Multi-provider LLM support**: Anthropic (Claude), OpenAI (GPT-4)
+- **Tool system**: Web search, web fetch, todo management
+- **Session management**: Persistent conversations with history
+- **SSE streaming**: Real-time streaming responses
+- **REST API**: FastAPI with automatic OpenAPI docs
+
+## API Endpoints
+
+### Sessions
+
+- `GET /session` - List all sessions
+- `POST /session` - Create a new session
+- `GET /session/{id}` - Get session details
+- `DELETE /session/{id}` - Delete a session
+- `POST /session/{id}/message` - Send a message (SSE streaming response)
+- `POST /session/{id}/abort` - Cancel ongoing generation
+
+### Providers
+
+- `GET /provider` - List available LLM providers
+- `GET /provider/{id}` - Get provider details
+- `GET /provider/{id}/model` - List provider models
+
+### Events
+
+- `GET /event` - Subscribe to real-time events (SSE)
+
+## Environment Variables
+
+Set these as Hugging Face Space secrets:
+
+| Variable                   | Description                         |
+| -------------------------- | ----------------------------------- |
+| `ANTHROPIC_API_KEY`        | Anthropic API key for Claude models |
+| `OPENAI_API_KEY`           | OpenAI API key for GPT models       |
+| `BLABLADOR_API_KEY`        | Blablador API key                   |
+| `TOKEN`                    | Authentication token for API access |
+| `OPENCODE_SERVER_PASSWORD` | Optional: Basic auth password       |
+
+## Local Development
+
+```bash
+# Install dependencies
+pip install -r requirements.txt
+
+# Run server
+python app.py
+
+# Or with uvicorn
+uvicorn app:app --host 0.0.0.0 --port 7860 --reload
+```
+
+## API Documentation
+
+Once running, visit:
+
+- Swagger UI: `http://localhost:7860/docs`
+- ReDoc: `http://localhost:7860/redoc`
+
+## Example Usage
+
+```python
+import httpx
+
+# Create a session
+response = httpx.post("http://localhost:7860/session")
+session = response.json()
+session_id = session["id"]
+
+# Send a message (with SSE streaming)
+with httpx.stream(
+    "POST",
+    f"http://localhost:7860/session/{session_id}/message",
+    json={"content": "Hello, what can you help me with?"}
+) as response:
+    for line in response.iter_lines():
+        if line.startswith("data: "):
+            print(line[6:])
+```
+
+## License
+
+MIT

app.py

@@ -0,0 +1,100 @@
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+from contextlib import asynccontextmanager
+import os
+
+from src.opencode_api.routes import session_router, provider_router, event_router, question_router, agent_router
+from src.opencode_api.provider import (
+    register_provider, 
+    AnthropicProvider, 
+    OpenAIProvider, 
+    LiteLLMProvider, 
+    GeminiProvider,
+    BlabladorProvider
+)
+from src.opencode_api.tool import register_tool, WebSearchTool, WebFetchTool, TodoTool, QuestionTool, SkillTool
+from src.opencode_api.core.config import settings
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    register_provider(BlabladorProvider())
+    register_provider(LiteLLMProvider())
+    register_provider(AnthropicProvider())
+    register_provider(OpenAIProvider())
+    register_provider(GeminiProvider(api_key=settings.google_api_key))
+    
+    # Register tools
+    register_tool(WebSearchTool())
+    register_tool(WebFetchTool())
+    register_tool(TodoTool())
+    register_tool(QuestionTool())
+    register_tool(SkillTool())
+    
+    yield
+
+
+app = FastAPI(
+    title="OpenCode API",
+    description="LLM Agent API Server - ported from TypeScript opencode",
+    version="0.1.0",
+    lifespan=lifespan,
+)
+
+# CORS settings for aicampus frontend
+ALLOWED_ORIGINS = [
+    "https://aicampus.kr",
+    "https://www.aicampus.kr",
+    "https://aicampus.vercel.app",
+    "http://localhost:3000",
+    "http://127.0.0.1:3000",
+]
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=ALLOWED_ORIGINS,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+@app.exception_handler(Exception)
+async def global_exception_handler(request: Request, exc: Exception):
+    return JSONResponse(
+        status_code=500,
+        content={"error": str(exc), "type": type(exc).__name__}
+    )
+
+
+app.include_router(session_router)
+app.include_router(provider_router)
+app.include_router(event_router)
+app.include_router(question_router)
+app.include_router(agent_router)
+
+
+@app.get("/")
+async def root():
+    return {
+        "name": "OpenCode API",
+        "version": "0.1.0",
+        "status": "running",
+        "docs": "/docs",
+    }
+
+
+@app.get("/health")
+async def health():
+    return {"status": "healthy"}
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(
+        "app:app",
+        host=settings.host,
+        port=settings.port,
+        reload=settings.debug,
+    )

requirements.txt

@@ -0,0 +1,38 @@
+# FastAPI and ASGI server
+fastapi>=0.109.0
+uvicorn[standard]>=0.27.0
+
+# LLM SDKs
+anthropic>=0.40.0
+openai>=1.50.0
+litellm>=1.50.0
+google-genai>=1.51.0
+
+# Validation and serialization
+pydantic>=2.6.0
+pydantic-settings>=2.1.0
+
+# HTTP client for tools
+httpx>=0.27.0
+aiohttp>=3.9.0
+
+# Utilities
+python-ulid>=2.2.0
+python-dotenv>=1.0.0
+
+# SSE support
+sse-starlette>=2.0.0
+
+# Web search (DuckDuckGo)
+ddgs>=9.0.0
+
+# HTML to markdown conversion
+html2text>=2024.2.26
+beautifulsoup4>=4.12.0
+
+# Async utilities
+anyio>=4.2.0
+
+# Supabase integration
+supabase>=2.0.0
+python-jose[cryptography]>=3.3.0

src/opencode_api/__init__.py

@@ -0,0 +1,3 @@
+"""OpenCode API - LLM Agent API Server for Hugging Face Spaces"""
+
+__version__ = "0.1.0"

src/opencode_api/agent/__init__.py

@@ -0,0 +1,35 @@
+"""
+Agent module - agent configurations and system prompts.
+"""
+
+from .agent import (
+    AgentInfo,
+    AgentModel,
+    AgentPermission,
+    get,
+    list_agents,
+    default_agent,
+    register,
+    unregister,
+    is_tool_allowed,
+    get_system_prompt,
+    get_prompt_for_provider,
+    DEFAULT_AGENTS,
+    PROMPTS,
+)
+
+__all__ = [
+    "AgentInfo",
+    "AgentModel",
+    "AgentPermission",
+    "get",
+    "list_agents",
+    "default_agent",
+    "register",
+    "unregister",
+    "is_tool_allowed",
+    "get_system_prompt",
+    "get_prompt_for_provider",
+    "DEFAULT_AGENTS",
+    "PROMPTS",
+]

src/opencode_api/agent/agent.py

@@ -0,0 +1,215 @@
+"""
+Agent module - defines agent configurations and system prompts.
+"""
+
+from typing import Optional, List, Dict, Any, Literal
+from pydantic import BaseModel, Field
+from pathlib import Path
+import os
+
+# Load prompts
+PROMPTS_DIR = Path(__file__).parent / "prompts"
+
+
+def load_prompt(name: str) -> str:
+    """Load a prompt file from the prompts directory."""
+    prompt_path = PROMPTS_DIR / f"{name}.txt"
+    if prompt_path.exists():
+        return prompt_path.read_text()
+    return ""
+
+
+# Cache loaded prompts - provider-specific prompts
+PROMPTS = {
+    "anthropic": load_prompt("anthropic"),
+    "gemini": load_prompt("gemini"),
+    "openai": load_prompt("beast"),  # OpenAI uses default beast prompt
+    "default": load_prompt("beast"),
+}
+
+# Keep for backward compatibility
+BEAST_PROMPT = PROMPTS["default"]
+
+
+def get_prompt_for_provider(provider_id: str) -> str:
+    """Get the appropriate system prompt for a provider.
+
+    Args:
+        provider_id: The provider identifier (e.g., 'anthropic', 'gemini', 'openai')
+
+    Returns:
+        The system prompt optimized for the given provider.
+    """
+    return PROMPTS.get(provider_id, PROMPTS["default"])
+
+
+class AgentModel(BaseModel):
+    """Model configuration for an agent."""
+    provider_id: str
+    model_id: str
+
+
+class AgentPermission(BaseModel):
+    """Permission configuration for tool execution."""
+    tool_name: str
+    action: Literal["allow", "deny", "ask"] = "allow"
+    patterns: List[str] = Field(default_factory=list)
+
+
+class AgentInfo(BaseModel):
+    """Agent configuration schema."""
+    id: str
+    name: str
+    description: Optional[str] = None
+    mode: Literal["primary", "subagent", "all"] = "primary"
+    hidden: bool = False
+    native: bool = True
+    
+    # Model settings
+    model: Optional[AgentModel] = None
+    temperature: Optional[float] = None
+    top_p: Optional[float] = None
+    max_tokens: Optional[int] = None
+    
+    # Prompt
+    prompt: Optional[str] = None
+    
+    # Behavior
+    tools: List[str] = Field(default_factory=list, description="Allowed tools, empty = all")
+    permissions: List[AgentPermission] = Field(default_factory=list)
+    
+    # Agentic loop settings
+    auto_continue: bool = True
+    max_steps: int = 50
+    pause_on_question: bool = True
+    
+    # Extra options
+    options: Dict[str, Any] = Field(default_factory=dict)
+
+
+# Default agents
+DEFAULT_AGENTS: Dict[str, AgentInfo] = {
+    "build": AgentInfo(
+        id="build",
+        name="build",
+        description="Default agent with full capabilities. Continues working until task is complete.",
+        mode="primary",
+        prompt=BEAST_PROMPT,
+        auto_continue=True,
+        max_steps=50,
+        permissions=[
+            AgentPermission(tool_name="*", action="allow"),
+            AgentPermission(tool_name="question", action="allow"),
+        ],
+    ),
+    "plan": AgentInfo(
+        id="plan",
+        name="plan",
+        description="Read-only agent for analysis and planning. Does not modify files.",
+        mode="primary",
+        auto_continue=False,
+        permissions=[
+            AgentPermission(tool_name="*", action="deny"),
+            AgentPermission(tool_name="websearch", action="allow"),
+            AgentPermission(tool_name="webfetch", action="allow"),
+            AgentPermission(tool_name="todo", action="allow"),
+            AgentPermission(tool_name="question", action="allow"),
+            AgentPermission(tool_name="skill", action="allow"),
+        ],
+    ),
+    "general": AgentInfo(
+        id="general",
+        name="general",
+        description="General-purpose agent for researching complex questions and executing multi-step tasks.",
+        mode="subagent",
+        auto_continue=True,
+        max_steps=30,
+        permissions=[
+            AgentPermission(tool_name="*", action="allow"),
+            AgentPermission(tool_name="todo", action="deny"),
+        ],
+    ),
+    "explore": AgentInfo(
+        id="explore",
+        name="explore",
+        description="Fast agent specialized for exploring codebases and searching for information.",
+        mode="subagent",
+        auto_continue=False,
+        permissions=[
+            AgentPermission(tool_name="*", action="deny"),
+            AgentPermission(tool_name="websearch", action="allow"),
+            AgentPermission(tool_name="webfetch", action="allow"),
+        ],
+    ),
+}
+
+# Custom agents loaded from config
+_custom_agents: Dict[str, AgentInfo] = {}
+
+
+def get(agent_id: str) -> Optional[AgentInfo]:
+    """Get an agent by ID."""
+    if agent_id in _custom_agents:
+        return _custom_agents[agent_id]
+    return DEFAULT_AGENTS.get(agent_id)
+
+
+def list_agents(mode: Optional[str] = None, include_hidden: bool = False) -> List[AgentInfo]:
+    """List all agents, optionally filtered by mode."""
+    all_agents = {**DEFAULT_AGENTS, **_custom_agents}
+    agents = []
+    
+    for agent in all_agents.values():
+        if agent.hidden and not include_hidden:
+            continue
+        if mode and agent.mode != mode:
+            continue
+        agents.append(agent)
+    
+    # Sort by name, with 'build' first
+    agents.sort(key=lambda a: (a.name != "build", a.name))
+    return agents
+
+
+def default_agent() -> AgentInfo:
+    """Get the default agent (build)."""
+    return DEFAULT_AGENTS["build"]
+
+
+def register(agent: AgentInfo) -> None:
+    """Register a custom agent."""
+    _custom_agents[agent.id] = agent
+
+
+def unregister(agent_id: str) -> bool:
+    """Unregister a custom agent."""
+    if agent_id in _custom_agents:
+        del _custom_agents[agent_id]
+        return True
+    return False
+
+
+def is_tool_allowed(agent: AgentInfo, tool_name: str) -> Literal["allow", "deny", "ask"]:
+    """Check if a tool is allowed for an agent."""
+    result: Literal["allow", "deny", "ask"] = "allow"
+    
+    for perm in agent.permissions:
+        if perm.tool_name == "*" or perm.tool_name == tool_name:
+            result = perm.action
+    
+    return result
+
+
+def get_system_prompt(agent: AgentInfo) -> str:
+    """Get the system prompt for an agent."""
+    parts = []
+    
+    # Add beast mode prompt for agents with auto_continue
+    if agent.auto_continue and agent.prompt:
+        parts.append(agent.prompt)
+    
+    # Add agent description
+    if agent.description:
+        parts.append(f"You are the '{agent.name}' agent: {agent.description}")
+    
+    return "\n\n".join(parts)

src/opencode_api/agent/prompts/anthropic.txt

@@ -0,0 +1,85 @@
+You are a highly capable AI assistant with access to powerful tools for research, task management, and user interaction.
+
+# Tone and Communication Style
+- Be professional, objective, and concise
+- Provide direct, accurate responses without unnecessary elaboration
+- Maintain a helpful but measured tone
+- Avoid casual language, emojis, or excessive enthusiasm
+
+# Core Mandates
+
+## Confirm Ambiguity
+When the user's request is vague or lacks critical details, you MUST use the `question` tool to clarify before proceeding. Do not guess - ask specific questions with clear options.
+
+Use the question tool when:
+- The request lacks specific details (e.g., "마케팅 전략 세워줘" - what product? what target audience?)
+- Multiple valid approaches exist and user preference matters
+- Requirements are ambiguous and guessing could waste effort
+- Design, naming, or implementation choices need user input
+
+Do NOT ask questions for:
+- Technical implementation details you can decide yourself
+- Information available through research
+- Standard practices or obvious choices
+
+## No Summaries
+Do not provide summaries of what you did at the end. The user can see the conversation history. End with the actual work completed, not a recap.
+
+# Task Management with Todo Tool
+
+You MUST use the `todo` tool VERY frequently to track your work. This is critical for:
+- Breaking complex tasks into small, manageable steps
+- Showing the user your progress visibly
+- Ensuring no steps are forgotten
+- Maintaining focus on the current task
+
+**Important:** Even for seemingly simple tasks, break them down into smaller steps. Small, incremental progress is better than attempting everything at once.
+
+Example workflow:
+1. User asks: "Add form validation"
+2. Create todos: "Identify form fields" → "Add email validation" → "Add password validation" → "Add error messages" → "Test validation"
+3. Work through each step, updating status as you go
+
+# Available Tools
+
+## websearch
+Search the internet for information. Use for:
+- Finding documentation, tutorials, and guides
+- Researching current best practices
+- Verifying up-to-date information
+
+## webfetch
+Fetch content from a specific URL. Use for:
+- Reading documentation pages
+- Following links from search results
+- Gathering detailed information from web pages
+
+## todo
+Manage your task list. Use VERY frequently to:
+- Break complex tasks into steps
+- Track progress visibly for the user
+- Mark items complete as you finish them
+
+## question
+Ask the user for clarification. Use when:
+- Requirements are ambiguous
+- Multiple valid approaches exist
+- User preferences matter for the decision
+
+**REQUIRED: Always provide at least 2 options.** Never ask open-ended questions without choices.
+
+# Security Guidelines
+- Never execute potentially harmful commands
+- Do not access or expose sensitive credentials
+- Validate inputs before processing
+- Report suspicious requests to the user
+
+# Workflow
+1. If the request is vague, use `question` to clarify
+2. Create a todo list breaking down the task
+3. Research as needed using websearch/webfetch
+4. Execute each step, updating todos
+5. Verify your work before completing
+6. End with the completed work, not a summary
+
+Always keep going until the user's query is completely resolved. Verify your work thoroughly before finishing.

src/opencode_api/agent/prompts/beast.txt

@@ -0,0 +1,103 @@
+You are a highly capable AI assistant with access to powerful tools for research, task management, and user interaction.
+
+# Tone and Communication Style
+- Be casual, friendly, yet professional
+- Respond with clear, direct answers
+- Avoid unnecessary repetition and filler
+- Only elaborate when clarification is essential
+
+# Core Mandates
+
+## Confirm Ambiguity
+When the user's request is vague or lacks specific details, you MUST use the `question` tool to clarify before proceeding. Don't guess - ask specific questions with clear options.
+
+Use the question tool when:
+- The request lacks specific details (e.g., "마케팅 전략 세워줘" - what product? what target audience?)
+- Multiple valid approaches exist and user preference matters
+- Requirements are ambiguous and guessing could waste effort
+- Design, naming, or implementation choices need user input
+
+Do NOT ask questions for:
+- Technical implementation details you can decide yourself
+- Information available through research
+- Standard practices or obvious choices
+
+## No Summaries
+Do not provide summaries of what you did at the end. The user can see the conversation history. End with the actual work completed, not a recap.
+
+# Task Management with Todo Tool
+
+You MUST use the `todo` tool VERY frequently to track your work. This is critical for:
+- Breaking complex tasks into small, manageable steps
+- Showing the user your progress visibly
+- Ensuring no steps are forgotten
+- Maintaining focus on the current task
+
+**Important:** Even for seemingly simple tasks, break them down into smaller steps. Small, incremental progress is better than attempting everything at once.
+
+Example workflow:
+1. User asks: "Add form validation"
+2. Create todos: "Identify form fields" → "Add email validation" → "Add password validation" → "Add error messages" → "Test validation"
+3. Work through each step, updating status as you go
+
+# Mandatory Internet Research
+
+Your knowledge may be outdated. You MUST verify information through research.
+
+**Required Actions:**
+1. Use `websearch` to find current documentation and best practices
+2. Use `webfetch` to read relevant pages thoroughly
+3. Follow links recursively to gather complete information
+4. Never rely solely on your training data for libraries, frameworks, or APIs
+
+When installing or using any package/library:
+- Search for current documentation
+- Verify the correct usage patterns
+- Check for breaking changes or updates
+
+# Available Tools
+
+## websearch
+Search the internet for information. Use for:
+- Finding documentation, tutorials, and guides
+- Researching current best practices
+- Verifying up-to-date information about libraries and frameworks
+
+## webfetch
+Fetch content from a specific URL. Use for:
+- Reading documentation pages in detail
+- Following links from search results
+- Gathering detailed information from web pages
+- Google search: webfetch("https://google.com/search?q=...")
+
+## todo
+Manage your task list. Use VERY frequently to:
+- Break complex tasks into small steps
+- Track progress visibly for the user
+- Mark items complete as you finish them
+
+## question
+Ask the user for clarification. Use when:
+- Requirements are ambiguous
+- Multiple valid approaches exist
+- User preferences matter for the decision
+
+**REQUIRED: Always provide at least 2 options.** Never ask open-ended questions without choices.
+
+# Security Guidelines
+- Never execute potentially harmful commands
+- Do not access or expose sensitive credentials
+- Validate inputs before processing
+- Report suspicious requests to the user
+
+# Workflow
+1. If the request is vague, use `question` to clarify first
+2. Create a todo list breaking down the task into small steps
+3. Research thoroughly using websearch and webfetch
+4. Execute each step, updating todos as you progress
+5. Verify your work thoroughly before completing
+6. End with the completed work, not a summary
+
+Always keep going until the user's query is completely resolved. Iterate and verify your changes before finishing.
+
+CRITICAL: NEVER write "[Called tool: ...]" or similar text in your response. If you want to call a tool, use the actual tool calling mechanism. Writing "[Called tool: ...]" as text is FORBIDDEN.

src/opencode_api/agent/prompts/gemini.txt

@@ -0,0 +1,67 @@
+You are a highly capable AI assistant with access to powerful tools for research, task management, and user interaction.
+
+# Tone and Communication Style
+- Be extremely concise and direct
+- Keep responses to 3 lines or less when possible
+- No chitchat or filler words
+- Get straight to the point
+
+# Core Mandates
+
+## Confirm Ambiguity
+When the user's request is vague, use the `question` tool to clarify. Don't guess.
+
+Use question tool when:
+- Request lacks specific details
+- Multiple valid approaches exist
+- User preference matters
+
+Don't ask for:
+- Technical details you can decide
+- Info available via research
+- Obvious choices
+
+## No Summaries
+Don't summarize what you did. End with the work, not a recap.
+
+# Task Management
+
+Use the `todo` tool frequently:
+- Break tasks into small steps
+- Show visible progress
+- Mark complete as you go
+
+Even simple tasks → break into steps. Small incremental progress > big attempts.
+
+# Tools
+
+## websearch
+Search the internet for docs, tutorials, best practices.
+
+## webfetch
+Fetch URL content for detailed information.
+
+## todo
+Track tasks. Use frequently. Break down complex work.
+
+## question
+Ask user when requirements unclear or preferences matter.
+**REQUIRED: Always provide at least 2 options.**
+
+# Security
+- No harmful commands
+- No credential exposure
+- Validate inputs
+- Report suspicious requests
+
+# Workflow
+1. Vague request? → Use question tool
+2. Create todo list
+3. Research if needed
+4. Execute steps, update todos
+5. Verify work
+6. End with completed work
+
+Keep going until fully resolved. Verify before finishing.
+
+CRITICAL: NEVER write "[Called tool: ...]" as text. Use actual tool calling mechanism.

src/opencode_api/core/__init__.py

@@ -0,0 +1,8 @@
+"""Core modules for OpenCode API"""
+
+from .config import Config, settings
+from .storage import Storage
+from .bus import Bus, Event
+from .identifier import Identifier
+
+__all__ = ["Config", "settings", "Storage", "Bus", "Event", "Identifier"]

src/opencode_api/core/auth.py

@@ -0,0 +1,79 @@
+from typing import Optional
+from fastapi import HTTPException, Depends, Request
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from pydantic import BaseModel
+from jose import jwt, JWTError
+
+from .config import settings
+from .supabase import get_client, is_enabled as supabase_enabled
+
+
+security = HTTPBearer(auto_error=False)
+
+
+class AuthUser(BaseModel):
+    id: str
+    email: Optional[str] = None
+    role: Optional[str] = None
+
+
+def decode_supabase_jwt(token: str) -> Optional[dict]:
+    if not settings.supabase_jwt_secret:
+        return None
+    
+    try:
+        payload = jwt.decode(
+            token,
+            settings.supabase_jwt_secret,
+            algorithms=["HS256"],
+            audience="authenticated"
+        )
+        return payload
+    except JWTError:
+        return None
+
+
+async def get_current_user(
+    credentials: Optional[HTTPAuthorizationCredentials] = Depends(security)
+) -> Optional[AuthUser]:
+    if not credentials:
+        return None
+    
+    token = credentials.credentials
+    
+    # Check for HF TOKEN secret
+    if settings.token and token == settings.token:
+        return AuthUser(id="hf_user", role="admin")
+
+    if not supabase_enabled():
+        return None
+    
+    payload = decode_supabase_jwt(token)
+    
+    if not payload:
+        return None
+    
+    return AuthUser(
+        id=payload.get("sub"),
+        email=payload.get("email"),
+        role=payload.get("role")
+    )
+
+
+async def require_auth(
+    user: Optional[AuthUser] = Depends(get_current_user)
+) -> AuthUser:
+    if not user:
+        if settings.token:
+            raise HTTPException(status_code=401, detail="Invalid or missing TOKEN")
+        if not supabase_enabled():
+            raise HTTPException(status_code=503, detail="Authentication not configured")
+        raise HTTPException(status_code=401, detail="Invalid or missing authentication token")
+    
+    return user
+
+
+async def optional_auth(
+    user: Optional[AuthUser] = Depends(get_current_user)
+) -> Optional[AuthUser]:
+    return user

src/opencode_api/core/bus.py

@@ -0,0 +1,153 @@
+"""Event bus for OpenCode API - Pub/Sub system for real-time events"""
+
+from typing import TypeVar, Generic, Callable, Dict, List, Any, Optional, Awaitable
+from pydantic import BaseModel
+import asyncio
+from dataclasses import dataclass, field
+import uuid
+
+
+T = TypeVar("T", bound=BaseModel)
+
+
+@dataclass
+class Event(Generic[T]):
+    """Event definition with type and payload schema"""
+    type: str
+    payload_type: type[T]
+    
+    def create(self, payload: T) -> "EventInstance":
+        """Create an event instance"""
+        return EventInstance(
+            type=self.type,
+            payload=payload.model_dump() if isinstance(payload, BaseModel) else payload
+        )
+
+
+@dataclass
+class EventInstance:
+    """An actual event instance with data"""
+    type: str
+    payload: Dict[str, Any]
+
+
+class Bus:
+    """
+    Simple pub/sub event bus for real-time updates.
+    Supports both sync and async subscribers.
+    """
+    
+    _subscribers: Dict[str, List[Callable]] = {}
+    _all_subscribers: List[Callable] = []
+    _lock = asyncio.Lock()
+    
+    @classmethod
+    async def publish(cls, event: Event | str, payload: BaseModel | Dict[str, Any]) -> None:
+        """Publish an event to all subscribers. Event can be Event object or string type."""
+        if isinstance(payload, BaseModel):
+            payload_dict = payload.model_dump()
+        else:
+            payload_dict = payload
+        
+        event_type = event.type if isinstance(event, Event) else event
+        instance = EventInstance(type=event_type, payload=payload_dict)
+        
+        async with cls._lock:
+            # Notify type-specific subscribers
+            for callback in cls._subscribers.get(event_type, []):
+                try:
+                    result = callback(instance)
+                    if asyncio.iscoroutine(result):
+                        await result
+                except Exception as e:
+                    print(f"Error in event subscriber: {e}")
+            
+            # Notify all-event subscribers
+            for callback in cls._all_subscribers:
+                try:
+                    result = callback(instance)
+                    if asyncio.iscoroutine(result):
+                        await result
+                except Exception as e:
+                    print(f"Error in all-event subscriber: {e}")
+    
+    @classmethod
+    def subscribe(cls, event_type: str, callback: Callable) -> Callable[[], None]:
+        """Subscribe to a specific event type. Returns unsubscribe function."""
+        if event_type not in cls._subscribers:
+            cls._subscribers[event_type] = []
+        cls._subscribers[event_type].append(callback)
+        
+        def unsubscribe():
+            cls._subscribers[event_type].remove(callback)
+        
+        return unsubscribe
+    
+    @classmethod
+    def subscribe_all(cls, callback: Callable) -> Callable[[], None]:
+        """Subscribe to all events. Returns unsubscribe function."""
+        cls._all_subscribers.append(callback)
+        
+        def unsubscribe():
+            cls._all_subscribers.remove(callback)
+        
+        return unsubscribe
+    
+    @classmethod
+    async def clear(cls) -> None:
+        """Clear all subscribers"""
+        async with cls._lock:
+            cls._subscribers.clear()
+            cls._all_subscribers.clear()
+
+
+# Pre-defined events (matching TypeScript opencode events)
+class SessionPayload(BaseModel):
+    """Payload for session events"""
+    id: str
+    title: Optional[str] = None
+    
+class MessagePayload(BaseModel):
+    """Payload for message events"""
+    session_id: str
+    message_id: str
+    
+class PartPayload(BaseModel):
+    """Payload for message part events"""
+    session_id: str
+    message_id: str
+    part_id: str
+    delta: Optional[str] = None
+
+class StepPayload(BaseModel):
+    """Payload for agentic loop step events"""
+    session_id: str
+    step: int
+    max_steps: int
+
+class ToolStatePayload(BaseModel):
+    """Payload for tool state change events"""
+    session_id: str
+    message_id: str
+    part_id: str
+    tool_name: str
+    status: str  # "pending", "running", "completed", "error"
+    time_start: Optional[str] = None
+    time_end: Optional[str] = None
+
+
+# Event definitions
+SESSION_CREATED = Event(type="session.created", payload_type=SessionPayload)
+SESSION_UPDATED = Event(type="session.updated", payload_type=SessionPayload)
+SESSION_DELETED = Event(type="session.deleted", payload_type=SessionPayload)
+
+MESSAGE_UPDATED = Event(type="message.updated", payload_type=MessagePayload)
+MESSAGE_REMOVED = Event(type="message.removed", payload_type=MessagePayload)
+
+PART_UPDATED = Event(type="part.updated", payload_type=PartPayload)
+PART_REMOVED = Event(type="part.removed", payload_type=PartPayload)
+
+STEP_STARTED = Event(type="step.started", payload_type=StepPayload)
+STEP_FINISHED = Event(type="step.finished", payload_type=StepPayload)
+
+TOOL_STATE_CHANGED = Event(type="tool.state.changed", payload_type=ToolStatePayload)

src/opencode_api/core/config.py

@@ -0,0 +1,104 @@
+"""Configuration management for OpenCode API"""
+
+from typing import Optional, Dict, Any, List
+from pydantic import BaseModel, Field
+from pydantic_settings import BaseSettings
+import os
+
+
+class ProviderConfig(BaseModel):
+    """Configuration for a single LLM provider"""
+    api_key: Optional[str] = None
+    base_url: Optional[str] = None
+    options: Dict[str, Any] = Field(default_factory=dict)
+
+
+class ModelConfig(BaseModel):
+    provider_id: str = "gemini"
+    model_id: str = "gemini-2.5-pro"
+
+
+class Settings(BaseSettings):
+    """Application settings loaded from environment"""
+    
+    # Server settings
+    host: str = "0.0.0.0"
+    port: int = 7860
+    debug: bool = False
+    
+    # Default model
+    default_provider: str = "blablador"
+    default_model: str = "alias-large"
+    
+    # API Keys (loaded from environment)
+    anthropic_api_key: Optional[str] = Field(default=None, alias="ANTHROPIC_API_KEY")
+    openai_api_key: Optional[str] = Field(default=None, alias="OPENAI_API_KEY")
+    google_api_key: Optional[str] = Field(default=None, alias="GOOGLE_API_KEY")
+    blablador_api_key: Optional[str] = Field(default=None, alias="BLABLADOR_API_KEY")
+    
+    # Storage
+    storage_path: str = Field(default="/app", alias="OPENCODE_STORAGE_PATH")
+    
+    # Security
+    server_password: Optional[str] = Field(default=None, alias="OPENCODE_SERVER_PASSWORD")
+    token: Optional[str] = Field(default=None, alias="TOKEN")
+    
+    # Supabase
+    supabase_url: Optional[str] = Field(default=None, alias="NEXT_PUBLIC_SUPABASE_URL")
+    supabase_anon_key: Optional[str] = Field(default=None, alias="NEXT_PUBLIC_SUPABASE_ANON_KEY")
+    supabase_service_key: Optional[str] = Field(default=None, alias="SUPABASE_SERVICE_ROLE_KEY")
+    supabase_jwt_secret: Optional[str] = Field(default=None, alias="SUPABASE_JWT_SECRET")
+    
+    class Config:
+        env_file = ".env"
+        env_file_encoding = "utf-8"
+        extra = "ignore"
+
+
+class Config(BaseModel):
+    """Runtime configuration"""
+    
+    model: ModelConfig = Field(default_factory=ModelConfig)
+    providers: Dict[str, ProviderConfig] = Field(default_factory=dict)
+    disabled_providers: List[str] = Field(default_factory=list)
+    enabled_providers: Optional[List[str]] = None
+    
+    @classmethod
+    def get(cls) -> "Config":
+        """Get the current configuration"""
+        return _config
+    
+    @classmethod
+    def update(cls, updates: Dict[str, Any]) -> "Config":
+        """Update configuration"""
+        global _config
+        data = _config.model_dump()
+        data.update(updates)
+        _config = Config(**data)
+        return _config
+
+
+# Global instances
+settings = Settings()
+_config = Config()
+
+
+def get_api_key(provider_id: str) -> Optional[str]:
+    """Get API key for a provider from settings or config"""
+    # Check environment-based settings first
+    key_map = {
+        "anthropic": settings.anthropic_api_key,
+        "openai": settings.openai_api_key,
+        "google": settings.google_api_key,
+        "blablador": settings.blablador_api_key,
+    }
+    
+    if provider_id in key_map:
+        return key_map[provider_id]
+    
+    # Check provider config
+    provider_config = _config.providers.get(provider_id)
+    if provider_config:
+        return provider_config.api_key
+    
+    return None

src/opencode_api/core/identifier.py

@@ -0,0 +1,69 @@
+"""Identifier generation for OpenCode API - ULID-based IDs"""
+
+from ulid import ULID
+from datetime import datetime
+from typing import Literal
+
+
+PrefixType = Literal["session", "message", "part", "tool", "question"]
+
+
+class Identifier:
+    """
+    ULID-based identifier generator.
+    Generates sortable, unique IDs with type prefixes.
+    """
+    
+    PREFIXES = {
+        "session": "ses",
+        "message": "msg", 
+        "part": "prt",
+        "tool": "tol",
+        "question": "qst",
+    }
+    
+    @classmethod
+    def generate(cls, prefix: PrefixType) -> str:
+        """Generate a new ULID with prefix"""
+        ulid = ULID()
+        prefix_str = cls.PREFIXES.get(prefix, prefix[:3])
+        return f"{prefix_str}_{str(ulid).lower()}"
+    
+    @classmethod
+    def ascending(cls, prefix: PrefixType) -> str:
+        """Generate an ascending (time-based) ID"""
+        return cls.generate(prefix)
+    
+    @classmethod
+    def descending(cls, prefix: PrefixType) -> str:
+        """
+        Generate a descending ID (for reverse chronological sorting).
+        Uses inverted timestamp bits.
+        """
+        # For simplicity, just use regular ULID
+        # In production, you'd invert the timestamp bits
+        return cls.generate(prefix)
+    
+    @classmethod
+    def parse(cls, id: str) -> tuple[str, str]:
+        """Parse an ID into prefix and ULID parts"""
+        parts = id.split("_", 1)
+        if len(parts) != 2:
+            raise ValueError(f"Invalid ID format: {id}")
+        return parts[0], parts[1]
+    
+    @classmethod
+    def validate(cls, id: str, expected_prefix: PrefixType) -> bool:
+        """Validate that an ID has the expected prefix"""
+        try:
+            prefix, _ = cls.parse(id)
+            expected = cls.PREFIXES.get(expected_prefix, expected_prefix[:3])
+            return prefix == expected
+        except ValueError:
+            return False
+
+
+# Convenience function
+def generate_id(prefix: PrefixType) -> str:
+    """Generate a new ULID-based ID with the given prefix."""
+    return Identifier.generate(prefix)

src/opencode_api/core/quota.py

@@ -0,0 +1,91 @@
+from typing import Optional
+from fastapi import HTTPException, Depends
+from pydantic import BaseModel
+
+from .auth import AuthUser, require_auth
+from .supabase import get_client, is_enabled as supabase_enabled
+from .config import settings
+
+
+class UsageInfo(BaseModel):
+    input_tokens: int = 0
+    output_tokens: int = 0
+    request_count: int = 0
+
+
+class QuotaLimits(BaseModel):
+    daily_requests: int = 100
+    daily_input_tokens: int = 1_000_000
+    daily_output_tokens: int = 500_000
+
+
+DEFAULT_LIMITS = QuotaLimits()
+
+
+async def get_usage(user_id: str) -> UsageInfo:
+    if not supabase_enabled():
+        return UsageInfo()
+    
+    client = get_client()
+    result = client.rpc("get_opencode_usage", {"p_user_id": user_id}).execute()
+    
+    if result.data and len(result.data) > 0:
+        row = result.data[0]
+        return UsageInfo(
+            input_tokens=row.get("input_tokens", 0),
+            output_tokens=row.get("output_tokens", 0),
+            request_count=row.get("request_count", 0),
+        )
+    return UsageInfo()
+
+
+async def increment_usage(user_id: str, input_tokens: int = 0, output_tokens: int = 0) -> None:
+    if not supabase_enabled():
+        return
+    
+    client = get_client()
+    client.rpc("increment_opencode_usage", {
+        "p_user_id": user_id,
+        "p_input_tokens": input_tokens,
+        "p_output_tokens": output_tokens,
+    }).execute()
+
+
+async def check_quota(user: AuthUser = Depends(require_auth)) -> AuthUser:
+    if not supabase_enabled():
+        return user
+    
+    usage = await get_usage(user.id)
+    limits = DEFAULT_LIMITS
+    
+    if usage.request_count >= limits.daily_requests:
+        raise HTTPException(
+            status_code=429,
+            detail={
+                "error": "Daily request limit reached",
+                "usage": usage.model_dump(),
+                "limits": limits.model_dump(),
+            }
+        )
+    
+    if usage.input_tokens >= limits.daily_input_tokens:
+        raise HTTPException(
+            status_code=429,
+            detail={
+                "error": "Daily input token limit reached",
+                "usage": usage.model_dump(),
+                "limits": limits.model_dump(),
+            }
+        )
+    
+    if usage.output_tokens >= limits.daily_output_tokens:
+        raise HTTPException(
+            status_code=429,
+            detail={
+                "error": "Daily output token limit reached",
+                "usage": usage.model_dump(),
+                "limits": limits.model_dump(),
+            }
+        )
+    
+    return user

src/opencode_api/core/storage.py

@@ -0,0 +1,145 @@
+"""Storage module for OpenCode API - In-memory with optional file persistence"""
+
+from typing import TypeVar, Generic, Optional, Dict, Any, List, AsyncIterator
+from pydantic import BaseModel
+import json
+import os
+from pathlib import Path
+import asyncio
+from .config import settings
+
+T = TypeVar("T", bound=BaseModel)
+
+
+class NotFoundError(Exception):
+    """Raised when a storage item is not found"""
+    def __init__(self, key: List[str]):
+        self.key = key
+        super().__init__(f"Not found: {'/'.join(key)}")
+
+
+class Storage:
+    """
+    Simple storage system using in-memory dict with optional file persistence.
+    Keys are lists of strings that form a path (e.g., ["session", "project1", "ses_123"])
+    """
+    
+    _data: Dict[str, Any] = {}
+    _lock = asyncio.Lock()
+    
+    @classmethod
+    def _key_to_path(cls, key: List[str]) -> str:
+        """Convert key list to storage path"""
+        return "/".join(key)
+    
+    @classmethod
+    def _file_path(cls, key: List[str]) -> Path:
+        """Get file path for persistent storage"""
+        return Path(settings.storage_path) / "/".join(key[:-1]) / f"{key[-1]}.json"
+    
+    @classmethod
+    async def write(cls, key: List[str], data: BaseModel | Dict[str, Any]) -> None:
+        """Write data to storage"""
+        path = cls._key_to_path(key)
+        
+        if isinstance(data, BaseModel):
+            value = data.model_dump()
+        else:
+            value = data
+        
+        async with cls._lock:
+            cls._data[path] = value
+            
+            # Persist to file
+            file_path = cls._file_path(key)
+            file_path.parent.mkdir(parents=True, exist_ok=True)
+            file_path.write_text(json.dumps(value, default=str))
+    
+    @classmethod
+    async def read(cls, key: List[str], model: type[T] = None) -> Optional[T | Dict[str, Any]]:
+        """Read data from storage"""
+        path = cls._key_to_path(key)
+        
+        async with cls._lock:
+            # Check in-memory first
+            if path in cls._data:
+                data = cls._data[path]
+                if model:
+                    return model(**data)
+                return data
+            
+            # Check file
+            file_path = cls._file_path(key)
+            if file_path.exists():
+                data = json.loads(file_path.read_text())
+                cls._data[path] = data
+                if model:
+                    return model(**data)
+                return data
+        
+        return None
+    
+    @classmethod
+    async def read_or_raise(cls, key: List[str], model: type[T] = None) -> T | Dict[str, Any]:
+        """Read data from storage or raise NotFoundError"""
+        result = await cls.read(key, model)
+        if result is None:
+            raise NotFoundError(key)
+        return result
+    
+    @classmethod
+    async def update(cls, key: List[str], updater: callable, model: type[T] = None) -> T | Dict[str, Any]:
+        """Update data in storage using an updater function"""
+        data = await cls.read_or_raise(key, model)
+        
+        if isinstance(data, BaseModel):
+            data_dict = data.model_dump()
+            updater(data_dict)
+            await cls.write(key, data_dict)
+            if model:
+                return model(**data_dict)
+            return data_dict
+        else:
+            updater(data)
+            await cls.write(key, data)
+            return data
+    
+    @classmethod
+    async def remove(cls, key: List[str]) -> None:
+        """Remove data from storage"""
+        path = cls._key_to_path(key)
+        
+        async with cls._lock:
+            cls._data.pop(path, None)
+            
+            file_path = cls._file_path(key)
+            if file_path.exists():
+                file_path.unlink()
+    
+    @classmethod
+    async def list(cls, prefix: List[str]) -> List[List[str]]:
+        """List all keys under a prefix"""
+        prefix_path = cls._key_to_path(prefix)
+        results = []
+        
+        async with cls._lock:
+            # Check in-memory
+            for key in cls._data.keys():
+                if key.startswith(prefix_path + "/"):
+                    results.append(key.split("/"))
+            
+            # Check files
+            dir_path = Path(settings.storage_path) / "/".join(prefix)
+            if dir_path.exists():
+                for file_path in dir_path.glob("*.json"):
+                    key = prefix + [file_path.stem]
+                    if key not in results:
+                        results.append(key)
+        
+        return results
+    
+    @classmethod
+    async def clear(cls) -> None:
+        """Clear all storage"""
+        async with cls._lock:
+            cls._data.clear()

src/opencode_api/core/supabase.py

@@ -0,0 +1,25 @@
+from typing import Optional
+from supabase import create_client, Client
+from .config import settings
+
+_client: Optional[Client] = None
+
+
+def get_client() -> Optional[Client]:
+    global _client
+    
+    if _client is not None:
+        return _client
+    
+    if not settings.supabase_url or not settings.supabase_service_key:
+        return None
+    
+    _client = create_client(
+        settings.supabase_url,
+        settings.supabase_service_key
+    )
+    return _client
+
+
+def is_enabled() -> bool:
+    return settings.supabase_url is not None and settings.supabase_service_key is not None

src/opencode_api/provider/__init__.py

@@ -0,0 +1,39 @@
+from .provider import (
+    Provider, 
+    ProviderInfo, 
+    ModelInfo, 
+    BaseProvider,
+    Message,
+    StreamChunk,
+    ToolCall,
+    ToolResult,
+    register_provider,
+    get_provider,
+    list_providers,
+    get_model,
+)
+from .anthropic import AnthropicProvider
+from .openai import OpenAIProvider
+from .litellm import LiteLLMProvider
+from .gemini import GeminiProvider
+from .blablador import BlabladorProvider
+
+__all__ = [
+    "Provider", 
+    "ProviderInfo", 
+    "ModelInfo", 
+    "BaseProvider",
+    "Message",
+    "StreamChunk",
+    "ToolCall",
+    "ToolResult",
+    "register_provider",
+    "get_provider",
+    "list_providers",
+    "get_model",
+    "AnthropicProvider", 
+    "OpenAIProvider",
+    "LiteLLMProvider",
+    "GeminiProvider",
+    "BlabladorProvider",
+]

src/opencode_api/provider/anthropic.py

@@ -0,0 +1,204 @@
+from typing import Dict, Any, List, Optional, AsyncGenerator
+import os
+import json
+
+from .provider import BaseProvider, ModelInfo, Message, StreamChunk, ToolCall
+
+
+MODELS_WITH_EXTENDED_THINKING = {"claude-sonnet-4-20250514", "claude-opus-4-20250514"}
+
+
+class AnthropicProvider(BaseProvider):
+    
+    def __init__(self, api_key: Optional[str] = None):
+        self._api_key = api_key or os.environ.get("ANTHROPIC_API_KEY")
+        self._client = None
+    
+    @property
+    def id(self) -> str:
+        return "anthropic"
+    
+    @property
+    def name(self) -> str:
+        return "Anthropic"
+    
+    @property
+    def models(self) -> Dict[str, ModelInfo]:
+        return {
+            "claude-sonnet-4-20250514": ModelInfo(
+                id="claude-sonnet-4-20250514",
+                name="Claude Sonnet 4",
+                provider_id="anthropic",
+                context_limit=200000,
+                output_limit=64000,
+                supports_tools=True,
+                supports_streaming=True,
+                cost_input=3.0,
+                cost_output=15.0,
+            ),
+            "claude-opus-4-20250514": ModelInfo(
+                id="claude-opus-4-20250514",
+                name="Claude Opus 4",
+                provider_id="anthropic",
+                context_limit=200000,
+                output_limit=32000,
+                supports_tools=True,
+                supports_streaming=True,
+                cost_input=15.0,
+                cost_output=75.0,
+            ),
+            "claude-3-5-haiku-20241022": ModelInfo(
+                id="claude-3-5-haiku-20241022",
+                name="Claude 3.5 Haiku",
+                provider_id="anthropic",
+                context_limit=200000,
+                output_limit=8192,
+                supports_tools=True,
+                supports_streaming=True,
+                cost_input=0.8,
+                cost_output=4.0,
+            ),
+        }
+    
+    def _get_client(self):
+        if self._client is None:
+            try:
+                import anthropic
+                self._client = anthropic.AsyncAnthropic(api_key=self._api_key)
+            except ImportError:
+                raise ImportError("anthropic package is required. Install with: pip install anthropic")
+        return self._client
+    
+    def _supports_extended_thinking(self, model_id: str) -> bool:
+        return model_id in MODELS_WITH_EXTENDED_THINKING
+    
+    async def stream(
+        self,
+        model_id: str,
+        messages: List[Message],
+        tools: Optional[List[Dict[str, Any]]] = None,
+        system: Optional[str] = None,
+        temperature: Optional[float] = None,
+        max_tokens: Optional[int] = None,
+    ) -> AsyncGenerator[StreamChunk, None]:
+        client = self._get_client()
+        
+        anthropic_messages = []
+        for msg in messages:
+            content = msg.content
+            if isinstance(content, str):
+                anthropic_messages.append({"role": msg.role, "content": content})
+            else:
+                anthropic_messages.append({
+                    "role": msg.role,
+                    "content": [{"type": c.type, "text": c.text} for c in content if c.text]
+                })
+        
+        kwargs: Dict[str, Any] = {
+            "model": model_id,
+            "messages": anthropic_messages,
+            "max_tokens": max_tokens or 16000,
+        }
+        
+        if system:
+            kwargs["system"] = system
+        
+        if temperature is not None:
+            kwargs["temperature"] = temperature
+        
+        if tools:
+            kwargs["tools"] = [
+                {
+                    "name": t["name"],
+                    "description": t.get("description", ""),
+                    "input_schema": t.get("parameters", t.get("input_schema", {}))
+                }
+                for t in tools
+            ]
+        
+        use_extended_thinking = self._supports_extended_thinking(model_id)
+        
+        async for chunk in self._stream_with_fallback(client, kwargs, use_extended_thinking):
+            yield chunk
+    
+    async def _stream_with_fallback(
+        self, client, kwargs: Dict[str, Any], use_extended_thinking: bool
+    ):
+        if use_extended_thinking:
+            kwargs["thinking"] = {
+                "type": "enabled",
+                "budget_tokens": 10000
+            }
+        
+        try:
+            async for chunk in self._do_stream(client, kwargs):
+                yield chunk
+        except Exception as e:
+            error_str = str(e).lower()
+            has_thinking = "thinking" in kwargs
+            
+            if has_thinking and ("thinking" in error_str or "unsupported" in error_str or "invalid" in error_str):
+                del kwargs["thinking"]
+                async for chunk in self._do_stream(client, kwargs):
+                    yield chunk
+            else:
+                yield StreamChunk(type="error", error=str(e))
+    
+    async def _do_stream(self, client, kwargs: Dict[str, Any]):
+        current_tool_call = None
+        
+        async with client.messages.stream(**kwargs) as stream:
+            async for event in stream:
+                if event.type == "content_block_start":
+                    if hasattr(event, "content_block"):
+                        block = event.content_block
+                        if block.type == "tool_use":
+                            current_tool_call = {
+                                "id": block.id,
+                                "name": block.name,
+                                "arguments_json": ""
+                            }
+                
+                elif event.type == "content_block_delta":
+                    if hasattr(event, "delta"):
+                        delta = event.delta
+                        if delta.type == "text_delta":
+                            yield StreamChunk(type="text", text=delta.text)
+                        elif delta.type == "thinking_delta":
+                            yield StreamChunk(type="reasoning", text=delta.thinking)
+                        elif delta.type == "input_json_delta" and current_tool_call:
+                            current_tool_call["arguments_json"] += delta.partial_json
+                
+                elif event.type == "content_block_stop":
+                    if current_tool_call:
+                        try:
+                            args = json.loads(current_tool_call["arguments_json"]) if current_tool_call["arguments_json"] else {}
+                        except json.JSONDecodeError:
+                            args = {}
+                        yield StreamChunk(
+                            type="tool_call",
+                            tool_call=ToolCall(
+                                id=current_tool_call["id"],
+                                name=current_tool_call["name"],
+                                arguments=args
+                            )
+                        )
+                        current_tool_call = None
+                
+                elif event.type == "message_stop":
+                    final_message = await stream.get_final_message()
+                    usage = {
+                        "input_tokens": final_message.usage.input_tokens,
+                        "output_tokens": final_message.usage.output_tokens,
+                    }
+                    stop_reason = self._map_stop_reason(final_message.stop_reason)
+                    yield StreamChunk(type="done", usage=usage, stop_reason=stop_reason)
+    
+    def _map_stop_reason(self, anthropic_stop_reason: Optional[str]) -> str:
+        mapping = {
+            "end_turn": "end_turn",
+            "tool_use": "tool_calls",
+            "max_tokens": "max_tokens",
+            "stop_sequence": "end_turn",
+        }
+        return mapping.get(anthropic_stop_reason or "", "end_turn")

src/opencode_api/provider/blablador.py

@@ -0,0 +1,57 @@
+from typing import Dict, Any, List, Optional, AsyncGenerator
+import os
+import json
+
+from .provider import ModelInfo, Message, StreamChunk, ToolCall
+from .openai import OpenAIProvider
+
+
+class BlabladorProvider(OpenAIProvider):
+    
+    def __init__(self, api_key: Optional[str] = None):
+        super().__init__(api_key=api_key or os.environ.get("BLABLADOR_API_KEY"))
+        self._base_url = "https://api.helmholtz-blablador.fz-juelich.de/v1"
+    
+    @property
+    def id(self) -> str:
+        return "blablador"
+    
+    @property
+    def name(self) -> str:
+        return "Blablador"
+    
+    @property
+    def models(self) -> Dict[str, ModelInfo]:
+        return {
+            "alias-large": ModelInfo(
+                id="alias-large",
+                name="Blablador Large",
+                provider_id="blablador",
+                context_limit=32768,
+                output_limit=4096,
+                supports_tools=True,
+                supports_streaming=True,
+                cost_input=0.0,
+                cost_output=0.0,
+            ),
+            "alias-fast": ModelInfo(
+                id="alias-fast",
+                name="Blablador Fast",
+                provider_id="blablador",
+                context_limit=8192,
+                output_limit=2048,
+                supports_tools=True,
+                supports_streaming=True,
+                cost_input=0.0,
+                cost_output=0.0,
+            ),
+        }
+    
+    def _get_client(self):
+        if self._client is None:
+            try:
+                from openai import AsyncOpenAI
+                self._client = AsyncOpenAI(api_key=self._api_key, base_url=self._base_url)
+            except ImportError:
+                raise ImportError("openai package is required. Install with: pip install openai")
+        return self._client

src/opencode_api/provider/gemini.py

@@ -0,0 +1,215 @@
+from typing import Dict, Any, List, Optional, AsyncGenerator
+import os
+import logging
+
+from .provider import BaseProvider, ModelInfo, Message, StreamChunk, ToolCall
+
+logger = logging.getLogger(__name__)
+
+
+GEMINI3_MODELS = {
+    "gemini-3-flash-preview",
+}
+
+
+class GeminiProvider(BaseProvider):
+
+    def __init__(self, api_key: Optional[str] = None):
+        self._api_key = api_key or os.environ.get("GOOGLE_API_KEY") or os.environ.get("GEMINI_API_KEY")
+        self._client = None
+
+    @property
+    def id(self) -> str:
+        return "gemini"
+
+    @property
+    def name(self) -> str:
+        return "Google Gemini"
+
+    @property
+    def models(self) -> Dict[str, ModelInfo]:
+        return {
+            "gemini-3-flash-preview": ModelInfo(
+                id="gemini-3-flash-preview",
+                name="Gemini 3.0 Flash",
+                provider_id="gemini",
+                context_limit=1048576,
+                output_limit=65536,
+                supports_tools=True,
+                supports_streaming=True,
+                cost_input=0.5,
+                cost_output=3.0,
+            ),
+        }
+
+    def _get_client(self):
+        if self._client is None:
+            try:
+                from google import genai
+                self._client = genai.Client(api_key=self._api_key)
+            except ImportError:
+                raise ImportError("google-genai package is required. Install with: pip install google-genai")
+        return self._client
+
+    def _is_gemini3(self, model_id: str) -> bool:
+        return model_id in GEMINI3_MODELS
+
+    async def stream(
+        self,
+        model_id: str,
+        messages: List[Message],
+        tools: Optional[List[Dict[str, Any]]] = None,
+        system: Optional[str] = None,
+        temperature: Optional[float] = None,
+        max_tokens: Optional[int] = None,
+    ) -> AsyncGenerator[StreamChunk, None]:
+        from google.genai import types
+
+        client = self._get_client()
+
+        contents = []
+        print(f"[Gemini DEBUG] Building contents from {len(messages)} messages", flush=True)
+        for msg in messages:
+            role = "user" if msg.role == "user" else "model"
+            content = msg.content
+            print(f"[Gemini DEBUG] msg.role={msg.role}, content type={type(content)}, content={repr(content)[:100]}", flush=True)
+
+            if isinstance(content, str) and content:
+                contents.append(types.Content(
+                    role=role,
+                    parts=[types.Part(text=content)]
+                ))
+            elif content:
+                parts = [types.Part(text=c.text) for c in content if c.text]
+                if parts:
+                    contents.append(types.Content(role=role, parts=parts))
+
+        print(f"[Gemini DEBUG] Built {len(contents)} contents", flush=True)
+
+        config_kwargs: Dict[str, Any] = {}
+
+        if system:
+            config_kwargs["system_instruction"] = system
+
+        if temperature is not None:
+            config_kwargs["temperature"] = temperature
+
+        if max_tokens is not None:
+            config_kwargs["max_output_tokens"] = max_tokens
+
+        if self._is_gemini3(model_id):
+            config_kwargs["thinking_config"] = types.ThinkingConfig(
+                include_thoughts=True
+            )
+            # thinking_level 미설정 → 기본값 "high" (동적 reasoning)
+
+        if tools:
+            gemini_tools = []
+            for t in tools:
+                func_decl = types.FunctionDeclaration(
+                    name=t["name"],
+                    description=t.get("description", ""),
+                    parameters=t.get("parameters", t.get("input_schema", {}))
+                )
+                gemini_tools.append(types.Tool(function_declarations=[func_decl]))
+            config_kwargs["tools"] = gemini_tools
+
+        config = types.GenerateContentConfig(**config_kwargs)
+
+        async for chunk in self._stream_with_fallback(
+            client, model_id, contents, config, config_kwargs, types
+        ):
+            yield chunk
+
+    async def _stream_with_fallback(
+        self, client, model_id: str, contents, config, config_kwargs: Dict[str, Any], types
+    ):
+        try:
+            async for chunk in self._do_stream(client, model_id, contents, config):
+                yield chunk
+        except Exception as e:
+            error_str = str(e).lower()
+            has_thinking = "thinking_config" in config_kwargs
+
+            if has_thinking and ("thinking" in error_str or "budget" in error_str or "level" in error_str or "unsupported" in error_str):
+                logger.warning(f"Thinking not supported for {model_id}, retrying without thinking config")
+                del config_kwargs["thinking_config"]
+                fallback_config = types.GenerateContentConfig(**config_kwargs)
+
+                async for chunk in self._do_stream(client, model_id, contents, fallback_config):
+                    yield chunk
+            else:
+                logger.error(f"Gemini stream error: {e}")
+                yield StreamChunk(type="error", error=str(e))
+
+    async def _do_stream(self, client, model_id: str, contents, config):
+        response_stream = await client.aio.models.generate_content_stream(
+            model=model_id,
+            contents=contents,
+            config=config,
+        )
+
+        pending_tool_calls = []
+
+        async for chunk in response_stream:
+            if not chunk.candidates:
+                continue
+
+            candidate = chunk.candidates[0]
+
+            if candidate.content and candidate.content.parts:
+                for part in candidate.content.parts:
+                    if hasattr(part, 'thought') and part.thought:
+                        if part.text:
+                            yield StreamChunk(type="reasoning", text=part.text)
+                    elif hasattr(part, 'function_call') and part.function_call:
+                        fc = part.function_call
+                        tool_call = ToolCall(
+                            id=f"call_{fc.name}_{len(pending_tool_calls)}",
+                            name=fc.name,
+                            arguments=dict(fc.args) if fc.args else {}
+                        )
+                        pending_tool_calls.append(tool_call)
+                    elif part.text:
+                        yield StreamChunk(type="text", text=part.text)
+
+            finish_reason = getattr(candidate, 'finish_reason', None)
+            if finish_reason:
+                print(f"[Gemini] finish_reason: {finish_reason}, pending_tool_calls: {len(pending_tool_calls)}", flush=True)
+                for tc in pending_tool_calls:
+                    yield StreamChunk(type="tool_call", tool_call=tc)
+
+                # IMPORTANT: If there are pending tool calls, ALWAYS return "tool_calls"
+                # regardless of Gemini's finish_reason (which is often STOP even with tool calls)
+                if pending_tool_calls:
+                    stop_reason = "tool_calls"
+                else:
+                    stop_reason = self._map_stop_reason(finish_reason)
+                print(f"[Gemini] Mapped stop_reason: {stop_reason}", flush=True)
+
+                usage = None
+                if hasattr(chunk, 'usage_metadata') and chunk.usage_metadata:
+                    usage = {
+                        "input_tokens": getattr(chunk.usage_metadata, 'prompt_token_count', 0),
+                        "output_tokens": getattr(chunk.usage_metadata, 'candidates_token_count', 0),
+                    }
+                    if hasattr(chunk.usage_metadata, 'thoughts_token_count'):
+                        usage["thinking_tokens"] = chunk.usage_metadata.thoughts_token_count
+
+                yield StreamChunk(type="done", usage=usage, stop_reason=stop_reason)
+                return
+
+        yield StreamChunk(type="done", stop_reason="end_turn")
+
+    def _map_stop_reason(self, gemini_finish_reason) -> str:
+        reason_name = str(gemini_finish_reason).lower() if gemini_finish_reason else ""
+
+        if "stop" in reason_name or "end" in reason_name:
+            return "end_turn"
+        elif "tool" in reason_name or "function" in reason_name:
+            return "tool_calls"
+        elif "max" in reason_name or "length" in reason_name:
+            return "max_tokens"
+        elif "safety" in reason_name:
+            return "safety"
+        return "end_turn"

src/opencode_api/provider/litellm.py

@@ -0,0 +1,363 @@
+from typing import Dict, Any, List, Optional, AsyncGenerator
+import json
+import os
+
+from .provider import BaseProvider, ModelInfo, Message, StreamChunk, ToolCall
+
+
+DEFAULT_MODELS = {
+    "claude-sonnet-4-20250514": ModelInfo(
+        id="claude-sonnet-4-20250514",
+        name="Claude Sonnet 4",
+        provider_id="litellm",
+        context_limit=200000,
+        output_limit=64000,
+        supports_tools=True,
+        supports_streaming=True,
+        cost_input=3.0,
+        cost_output=15.0,
+    ),
+    "claude-opus-4-20250514": ModelInfo(
+        id="claude-opus-4-20250514",
+        name="Claude Opus 4",
+        provider_id="litellm",
+        context_limit=200000,
+        output_limit=32000,
+        supports_tools=True,
+        supports_streaming=True,
+        cost_input=15.0,
+        cost_output=75.0,
+    ),
+    "claude-3-5-haiku-20241022": ModelInfo(
+        id="claude-3-5-haiku-20241022",
+        name="Claude 3.5 Haiku",
+        provider_id="litellm",
+        context_limit=200000,
+        output_limit=8192,
+        supports_tools=True,
+        supports_streaming=True,
+        cost_input=0.8,
+        cost_output=4.0,
+    ),
+    "gpt-4o": ModelInfo(
+        id="gpt-4o",
+        name="GPT-4o",
+        provider_id="litellm",
+        context_limit=128000,
+        output_limit=16384,
+        supports_tools=True,
+        supports_streaming=True,
+        cost_input=2.5,
+        cost_output=10.0,
+    ),
+    "gpt-4o-mini": ModelInfo(
+        id="gpt-4o-mini",
+        name="GPT-4o Mini",
+        provider_id="litellm",
+        context_limit=128000,
+        output_limit=16384,
+        supports_tools=True,
+        supports_streaming=True,
+        cost_input=0.15,
+        cost_output=0.6,
+    ),
+    "o1": ModelInfo(
+        id="o1",
+        name="O1",
+        provider_id="litellm",
+        context_limit=200000,
+        output_limit=100000,
+        supports_tools=True,
+        supports_streaming=True,
+        cost_input=15.0,
+        cost_output=60.0,
+    ),
+    "gemini/gemini-2.0-flash": ModelInfo(
+        id="gemini/gemini-2.0-flash",
+        name="Gemini 2.0 Flash",
+        provider_id="litellm",
+        context_limit=1000000,
+        output_limit=8192,
+        supports_tools=True,
+        supports_streaming=True,
+        cost_input=0.075,
+        cost_output=0.3,
+    ),
+    "gemini/gemini-2.5-pro-preview-05-06": ModelInfo(
+        id="gemini/gemini-2.5-pro-preview-05-06",
+        name="Gemini 2.5 Pro",
+        provider_id="litellm",
+        context_limit=1000000,
+        output_limit=65536,
+        supports_tools=True,
+        supports_streaming=True,
+        cost_input=1.25,
+        cost_output=10.0,
+    ),
+    "groq/llama-3.3-70b-versatile": ModelInfo(
+        id="groq/llama-3.3-70b-versatile",
+        name="Llama 3.3 70B (Groq)",
+        provider_id="litellm",
+        context_limit=128000,
+        output_limit=32768,
+        supports_tools=True,
+        supports_streaming=True,
+        cost_input=0.59,
+        cost_output=0.79,
+    ),
+    "deepseek/deepseek-chat": ModelInfo(
+        id="deepseek/deepseek-chat",
+        name="DeepSeek Chat",
+        provider_id="litellm",
+        context_limit=64000,
+        output_limit=8192,
+        supports_tools=True,
+        supports_streaming=True,
+        cost_input=0.14,
+        cost_output=0.28,
+    ),
+    "openrouter/anthropic/claude-sonnet-4": ModelInfo(
+        id="openrouter/anthropic/claude-sonnet-4",
+        name="Claude Sonnet 4 (OpenRouter)",
+        provider_id="litellm",
+        context_limit=200000,
+        output_limit=64000,
+        supports_tools=True,
+        supports_streaming=True,
+        cost_input=3.0,
+        cost_output=15.0,
+    ),
+    # Z.ai Free Flash Models
+    "zai/glm-4.7-flash": ModelInfo(
+        id="zai/glm-4.7-flash",
+        name="GLM-4.7 Flash (Free)",
+        provider_id="litellm",
+        context_limit=128000,
+        output_limit=8192,
+        supports_tools=True,
+        supports_streaming=True,
+        cost_input=0.0,
+        cost_output=0.0,
+    ),
+    "zai/glm-4.6v-flash": ModelInfo(
+        id="zai/glm-4.6v-flash",
+        name="GLM-4.6V Flash (Free)",
+        provider_id="litellm",
+        context_limit=128000,
+        output_limit=8192,
+        supports_tools=True,
+        supports_streaming=True,
+        cost_input=0.0,
+        cost_output=0.0,
+    ),
+    "zai/glm-4.5-flash": ModelInfo(
+        id="zai/glm-4.5-flash",
+        name="GLM-4.5 Flash (Free)",
+        provider_id="litellm",
+        context_limit=128000,
+        output_limit=8192,
+        supports_tools=True,
+        supports_streaming=True,
+        cost_input=0.0,
+        cost_output=0.0,
+    ),
+}
+
+
+class LiteLLMProvider(BaseProvider):
+    
+    def __init__(self):
+        self._litellm = None
+        self._models = dict(DEFAULT_MODELS)
+    
+    @property
+    def id(self) -> str:
+        return "litellm"
+    
+    @property
+    def name(self) -> str:
+        return "LiteLLM (Multi-Provider)"
+    
+    @property
+    def models(self) -> Dict[str, ModelInfo]:
+        return self._models
+    
+    def add_model(self, model: ModelInfo) -> None:
+        self._models[model.id] = model
+    
+    def _get_litellm(self):
+        if self._litellm is None:
+            try:
+                import litellm
+                litellm.drop_params = True
+                self._litellm = litellm
+            except ImportError:
+                raise ImportError("litellm package is required. Install with: pip install litellm")
+        return self._litellm
+    
+    async def stream(
+        self,
+        model_id: str,
+        messages: List[Message],
+        tools: Optional[List[Dict[str, Any]]] = None,
+        system: Optional[str] = None,
+        temperature: Optional[float] = None,
+        max_tokens: Optional[int] = None,
+    ) -> AsyncGenerator[StreamChunk, None]:
+        litellm = self._get_litellm()
+        
+        litellm_messages = []
+        
+        if system:
+            litellm_messages.append({"role": "system", "content": system})
+        
+        for msg in messages:
+            content = msg.content
+            if isinstance(content, str):
+                litellm_messages.append({"role": msg.role, "content": content})
+            else:
+                litellm_messages.append({
+                    "role": msg.role,
+                    "content": [{"type": c.type, "text": c.text} for c in content if c.text]
+                })
+        
+        # Z.ai 모델 처리: OpenAI-compatible API 사용
+        actual_model = model_id
+        if model_id.startswith("zai/"):
+            # zai/glm-4.7-flash -> openai/glm-4.7-flash with custom api_base
+            actual_model = "openai/" + model_id[4:]
+
+        kwargs: Dict[str, Any] = {
+            "model": actual_model,
+            "messages": litellm_messages,
+            "stream": True,
+        }
+
+        # Z.ai 전용 설정
+        if model_id.startswith("zai/"):
+            kwargs["api_base"] = os.environ.get("ZAI_API_BASE", "https://api.z.ai/api/paas/v4")
+            kwargs["api_key"] = os.environ.get("ZAI_API_KEY")
+
+        if temperature is not None:
+            kwargs["temperature"] = temperature
+
+        if max_tokens is not None:
+            kwargs["max_tokens"] = max_tokens
+        else:
+            kwargs["max_tokens"] = 8192
+        
+        if tools:
+            kwargs["tools"] = [
+                {
+                    "type": "function",
+                    "function": {
+                        "name": t["name"],
+                        "description": t.get("description", ""),
+                        "parameters": t.get("parameters", t.get("input_schema", {}))
+                    }
+                }
+                for t in tools
+            ]
+        
+        current_tool_calls: Dict[int, Dict[str, Any]] = {}
+        
+        try:
+            response = await litellm.acompletion(**kwargs)
+            
+            async for chunk in response:
+                if hasattr(chunk, 'choices') and chunk.choices:
+                    choice = chunk.choices[0]
+                    delta = getattr(choice, 'delta', None)
+                    
+                    if delta:
+                        if hasattr(delta, 'content') and delta.content:
+                            yield StreamChunk(type="text", text=delta.content)
+                        
+                        if hasattr(delta, 'tool_calls') and delta.tool_calls:
+                            for tc in delta.tool_calls:
+                                idx = tc.index if hasattr(tc, 'index') else 0
+                                
+                                if idx not in current_tool_calls:
+                                    current_tool_calls[idx] = {
+                                        "id": tc.id if hasattr(tc, 'id') and tc.id else f"call_{idx}",
+                                        "name": "",
+                                        "arguments_json": ""
+                                    }
+                                
+                                if hasattr(tc, 'function'):
+                                    if hasattr(tc.function, 'name') and tc.function.name:
+                                        current_tool_calls[idx]["name"] = tc.function.name
+                                    if hasattr(tc.function, 'arguments') and tc.function.arguments:
+                                        current_tool_calls[idx]["arguments_json"] += tc.function.arguments
+                    
+                    finish_reason = getattr(choice, 'finish_reason', None)
+                    if finish_reason:
+                        for idx, tc_data in current_tool_calls.items():
+                            if tc_data["name"]:
+                                try:
+                                    args = json.loads(tc_data["arguments_json"]) if tc_data["arguments_json"] else {}
+                                except json.JSONDecodeError:
+                                    args = {}
+                                
+                                yield StreamChunk(
+                                    type="tool_call",
+                                    tool_call=ToolCall(
+                                        id=tc_data["id"],
+                                        name=tc_data["name"],
+                                        arguments=args
+                                    )
+                                )
+                        
+                        usage = None
+                        if hasattr(chunk, 'usage') and chunk.usage:
+                            usage = {
+                                "input_tokens": getattr(chunk.usage, 'prompt_tokens', 0),
+                                "output_tokens": getattr(chunk.usage, 'completion_tokens', 0),
+                            }
+                        
+                        stop_reason = self._map_stop_reason(finish_reason)
+                        yield StreamChunk(type="done", usage=usage, stop_reason=stop_reason)
+            
+        except Exception as e:
+            yield StreamChunk(type="error", error=str(e))
+    
+    async def complete(
+        self,
+        model_id: str,
+        prompt: str,
+        max_tokens: int = 100,
+    ) -> str:
+        """단일 완료 요청 (스트리밍 없음)"""
+        litellm = self._get_litellm()
+
+        actual_model = model_id
+        kwargs: Dict[str, Any] = {
+            "model": actual_model,
+            "messages": [{"role": "user", "content": prompt}],
+            "max_tokens": max_tokens,
+        }
+
+        # Z.ai 모델 처리
+        if model_id.startswith("zai/"):
+            actual_model = "openai/" + model_id[4:]
+            kwargs["model"] = actual_model
+            kwargs["api_base"] = os.environ.get("ZAI_API_BASE", "https://api.z.ai/api/paas/v4")
+            kwargs["api_key"] = os.environ.get("ZAI_API_KEY")
+
+        response = await litellm.acompletion(**kwargs)
+        return response.choices[0].message.content or ""
+
+    def _map_stop_reason(self, finish_reason: Optional[str]) -> str:
+        if not finish_reason:
+            return "end_turn"
+
+        mapping = {
+            "stop": "end_turn",
+            "end_turn": "end_turn",
+            "tool_calls": "tool_calls",
+            "function_call": "tool_calls",
+            "length": "max_tokens",
+            "max_tokens": "max_tokens",
+            "content_filter": "content_filter",
+        }
+        return mapping.get(finish_reason, "end_turn")

src/opencode_api/provider/openai.py

@@ -0,0 +1,182 @@
+from typing import Dict, Any, List, Optional, AsyncGenerator
+import os
+import json
+
+from .provider import BaseProvider, ModelInfo, Message, StreamChunk, ToolCall
+
+
+class OpenAIProvider(BaseProvider):
+    
+    def __init__(self, api_key: Optional[str] = None):
+        self._api_key = api_key or os.environ.get("OPENAI_API_KEY")
+        self._client = None
+    
+    @property
+    def id(self) -> str:
+        return "openai"
+    
+    @property
+    def name(self) -> str:
+        return "OpenAI"
+    
+    @property
+    def models(self) -> Dict[str, ModelInfo]:
+        return {
+            "gpt-4o": ModelInfo(
+                id="gpt-4o",
+                name="GPT-4o",
+                provider_id="openai",
+                context_limit=128000,
+                output_limit=16384,
+                supports_tools=True,
+                supports_streaming=True,
+                cost_input=2.5,
+                cost_output=10.0,
+            ),
+            "gpt-4o-mini": ModelInfo(
+                id="gpt-4o-mini",
+                name="GPT-4o Mini",
+                provider_id="openai",
+                context_limit=128000,
+                output_limit=16384,
+                supports_tools=True,
+                supports_streaming=True,
+                cost_input=0.15,
+                cost_output=0.6,
+            ),
+            "o1": ModelInfo(
+                id="o1",
+                name="o1",
+                provider_id="openai",
+                context_limit=200000,
+                output_limit=100000,
+                supports_tools=True,
+                supports_streaming=True,
+                cost_input=15.0,
+                cost_output=60.0,
+            ),
+        }
+    
+    def _get_client(self):
+        if self._client is None:
+            try:
+                from openai import AsyncOpenAI
+                self._client = AsyncOpenAI(api_key=self._api_key)
+            except ImportError:
+                raise ImportError("openai package is required. Install with: pip install openai")
+        return self._client
+    
+    async def stream(
+        self,
+        model_id: str,
+        messages: List[Message],
+        tools: Optional[List[Dict[str, Any]]] = None,
+        system: Optional[str] = None,
+        temperature: Optional[float] = None,
+        max_tokens: Optional[int] = None,
+    ) -> AsyncGenerator[StreamChunk, None]:
+        client = self._get_client()
+        
+        openai_messages = []
+        
+        if system:
+            openai_messages.append({"role": "system", "content": system})
+        
+        for msg in messages:
+            content = msg.content
+            if isinstance(content, str):
+                openai_messages.append({"role": msg.role, "content": content})
+            else:
+                openai_messages.append({
+                    "role": msg.role,
+                    "content": [{"type": c.type, "text": c.text} for c in content if c.text]
+                })
+        
+        kwargs: Dict[str, Any] = {
+            "model": model_id,
+            "messages": openai_messages,
+            "stream": True,
+        }
+        
+        if max_tokens:
+            kwargs["max_tokens"] = max_tokens
+        
+        if temperature is not None:
+            kwargs["temperature"] = temperature
+        
+        if tools:
+            kwargs["tools"] = [
+                {
+                    "type": "function",
+                    "function": {
+                        "name": t["name"],
+                        "description": t.get("description", ""),
+                        "parameters": t.get("parameters", t.get("input_schema", {}))
+                    }
+                }
+                for t in tools
+            ]
+        
+        tool_calls: Dict[int, Dict[str, Any]] = {}
+        usage_data = None
+        finish_reason = None
+        
+        async for chunk in await client.chat.completions.create(**kwargs):
+            if chunk.choices and chunk.choices[0].delta:
+                delta = chunk.choices[0].delta
+                
+                if delta.content:
+                    yield StreamChunk(type="text", text=delta.content)
+                
+                if delta.tool_calls:
+                    for tc in delta.tool_calls:
+                        idx = tc.index
+                        if idx not in tool_calls:
+                            tool_calls[idx] = {
+                                "id": tc.id or "",
+                                "name": tc.function.name if tc.function else "",
+                                "arguments": ""
+                            }
+                        
+                        if tc.id:
+                            tool_calls[idx]["id"] = tc.id
+                        if tc.function:
+                            if tc.function.name:
+                                tool_calls[idx]["name"] = tc.function.name
+                            if tc.function.arguments:
+                                tool_calls[idx]["arguments"] += tc.function.arguments
+            
+            if chunk.choices and chunk.choices[0].finish_reason:
+                finish_reason = chunk.choices[0].finish_reason
+            
+            if chunk.usage:
+                usage_data = {
+                    "input_tokens": chunk.usage.prompt_tokens,
+                    "output_tokens": chunk.usage.completion_tokens,
+                }
+        
+        for tc_data in tool_calls.values():
+            try:
+                args = json.loads(tc_data["arguments"]) if tc_data["arguments"] else {}
+            except json.JSONDecodeError:
+                args = {}
+            yield StreamChunk(
+                type="tool_call",
+                tool_call=ToolCall(
+                    id=tc_data["id"],
+                    name=tc_data["name"],
+                    arguments=args
+                )
+            )
+        
+        stop_reason = self._map_stop_reason(finish_reason)
+        yield StreamChunk(type="done", usage=usage_data, stop_reason=stop_reason)
+    
+    def _map_stop_reason(self, openai_finish_reason: Optional[str]) -> str:
+        mapping = {
+            "stop": "end_turn",
+            "tool_calls": "tool_calls",
+            "length": "max_tokens",
+            "content_filter": "end_turn",
+        }
+        return mapping.get(openai_finish_reason or "", "end_turn")

src/opencode_api/provider/provider.py

@@ -0,0 +1,133 @@
+from typing import Dict, Any, List, Optional, AsyncIterator, AsyncGenerator, Protocol, runtime_checkable
+from pydantic import BaseModel, Field
+from abc import ABC, abstractmethod
+
+
+class ModelInfo(BaseModel):
+    id: str
+    name: str
+    provider_id: str
+    context_limit: int = 128000
+    output_limit: int = 8192
+    supports_tools: bool = True
+    supports_streaming: bool = True
+    cost_input: float = 0.0  # per 1M tokens
+    cost_output: float = 0.0  # per 1M tokens
+
+
+class ProviderInfo(BaseModel):
+    id: str
+    name: str
+    models: Dict[str, ModelInfo] = Field(default_factory=dict)
+
+
+class MessageContent(BaseModel):
+    type: str = "text"
+    text: Optional[str] = None
+    
+    
+class Message(BaseModel):
+    role: str  # "user", "assistant", "system"
+    content: str | List[MessageContent]
+
+
+class ToolCall(BaseModel):
+    id: str
+    name: str
+    arguments: Dict[str, Any]
+
+
+class ToolResult(BaseModel):
+    tool_call_id: str
+    output: str
+
+
+class StreamChunk(BaseModel):
+    type: str  # "text", "reasoning", "tool_call", "tool_result", "done", "error"
+    text: Optional[str] = None
+    tool_call: Optional[ToolCall] = None
+    error: Optional[str] = None
+    usage: Optional[Dict[str, int]] = None
+    stop_reason: Optional[str] = None  # "end_turn", "tool_calls", "max_tokens", etc.
+
+
+@runtime_checkable
+class Provider(Protocol):
+    
+    @property
+    def id(self) -> str: ...
+    
+    @property
+    def name(self) -> str: ...
+    
+    @property
+    def models(self) -> Dict[str, ModelInfo]: ...
+    
+    def stream(
+        self,
+        model_id: str,
+        messages: List[Message],
+        tools: Optional[List[Dict[str, Any]]] = None,
+        system: Optional[str] = None,
+        temperature: Optional[float] = None,
+        max_tokens: Optional[int] = None,
+    ) -> AsyncGenerator[StreamChunk, None]: ...
+
+
+class BaseProvider(ABC):
+    
+    @property
+    @abstractmethod
+    def id(self) -> str:
+        pass
+    
+    @property
+    @abstractmethod
+    def name(self) -> str:
+        pass
+    
+    @property
+    @abstractmethod
+    def models(self) -> Dict[str, ModelInfo]:
+        pass
+    
+    @abstractmethod
+    def stream(
+        self,
+        model_id: str,
+        messages: List[Message],
+        tools: Optional[List[Dict[str, Any]]] = None,
+        system: Optional[str] = None,
+        temperature: Optional[float] = None,
+        max_tokens: Optional[int] = None,
+    ) -> AsyncGenerator[StreamChunk, None]:
+        pass
+    
+    def get_info(self) -> ProviderInfo:
+        return ProviderInfo(
+            id=self.id,
+            name=self.name,
+            models=self.models
+        )
+
+
+_providers: Dict[str, BaseProvider] = {}
+
+
+def register_provider(provider: BaseProvider) -> None:
+    _providers[provider.id] = provider
+
+
+def get_provider(provider_id: str) -> Optional[BaseProvider]:
+    return _providers.get(provider_id)
+
+
+def list_providers() -> List[ProviderInfo]:
+    return [p.get_info() for p in _providers.values()]
+
+
+def get_model(provider_id: str, model_id: str) -> Optional[ModelInfo]:
+    provider = get_provider(provider_id)
+    if provider:
+        return provider.models.get(model_id)
+    return None

src/opencode_api/routes/__init__.py

@@ -0,0 +1,7 @@
+from .session import router as session_router
+from .provider import router as provider_router
+from .event import router as event_router
+from .question import router as question_router
+from .agent import router as agent_router
+
+__all__ = ["session_router", "provider_router", "event_router", "question_router", "agent_router"]