| name | opcode_api |
| description | Guide for Kimix opencode-style HTTP server (FastAPI + SSE), including route definitions, builtin endpoints, and app.post usage. |
Kimix Opencode API Server Guide
This guide explains the FastAPI-based HTTP server in kimix.server.app that exposes an opencode-compatible REST API with SSE event streaming.
Application Factory
The server is created via create_app() in kimix/server/app.py.
from kimix.server.app import create_app
app = create_app()
create_app() configures:
FastAPI(title="Kimix API", version="0.1.0", docs_url="/docs", openapi_url="/openapi.json", redoc_url="/redoc")CORSMiddleware with allow_origins=["*"]- Shutdown handler that wakes all SSE queues
Interface Function Format
Route handlers are defined inside create_app() using FastAPI decorator methods. The typical structure is:
@app.<method>(
"/path",
response_model=ResponseModel,
tags=["Tag"],
summary="Short title",
description="Longer explanation",
responses={404: {"model": ErrorResponse, "description": "Not found"}},
status_code=200,
)
async def handler_name(param: Type) -> ReturnType:
...
Example from app.py
@app.post(
"/session",
response_model=SessionResponse,
tags=["Session"],
summary="Create session",
description="Create a new chat session. Returns the session metadata.",
status_code=200,
)
async def create_session(body: CreateSessionRequest) -> Dict[str, Any]:
info = await session_manager.create_session(title=body.title)
return info.to_dict()
Handler style rules
- All route handlers are async.
- Path parameters use
{name} in the route string and are declared as function arguments (sessionID: str).
- Query parameters use
Optional[int] = Query(default=None, description="...").
- Request bodies use Pydantic models (
body: CreateSessionRequest).
- Errors raise
HTTPException(status_code=..., detail="...").
Builtin Endpoints (routes starting with /)
| Method | Path | Summary |
|---|
| GET | /global/health | Health check |
| GET | /event | SSE event stream (global) |
| POST | /session | Create session |
| GET | /session | List sessions |
| GET | /session/status | Get all session statuses |
| GET | /session/{sessionID} | Get session info |
| DELETE | /session/{sessionID} | Delete session |
| GET | /session/{sessionID}/message | Get messages |
| POST | /session/{sessionID}/prompt_async | Send message (fire-and-forget, 204) |
| POST | /session/{sessionID}/abort | Abort session |
| POST | /session/{sessionID}/permissions/{permissionID} | Grant permission |
SSE /event format
OpenCode protocol: no SSE event: field is used. All events are plain data: {json}\n\n lines.
Example initial event:
data: {"type": "server.connected", "properties": {}}\n\n
Heartbeat comment (no event: field):
: heartbeat\n\n
app.post Usage
app.post is used for state-changing operations. Common patterns:
1. Create resource (returns 200 + body)
@app.post(
"/session",
response_model=SessionResponse,
tags=["Session"],
summary="Create session",
status_code=200,
)
async def create_session(body: CreateSessionRequest) -> Dict[str, Any]:
info = await session_manager.create_session(title=body.title)
return info.to_dict()
2. Fire-and-forget (returns 204, no body)
@app.post(
"/session/{sessionID}/prompt_async",
status_code=204,
tags=["Message"],
summary="Send message (async)",
description="Send a prompt fire-and-forget style. Returns 204 immediately. Response events are streamed via SSE /event.",
responses={
404: {"model": ErrorResponse, "description": "Session not found"},
400: {"model": ErrorResponse, "description": "Invalid input"},
},
)
async def send_prompt_async(sessionID: str, body: PromptInput) -> Response:
text_parts = [p.text for p in body.parts if p.type == "text" and p.text]
text = "\n".join(text_parts)
if not text:
raise HTTPException(status_code=400, detail="No text content in parts")
try:
await session_manager.prompt_async(sessionID, text, agent=body.agent)
except KeyError:
raise HTTPException(status_code=404, detail=f"Session not found: {sessionID}")
return Response(status_code=204)
3. Action endpoint (returns 200 + empty body)
@app.post(
"/session/{sessionID}/abort",
tags=["Session"],
summary="Abort session",
description="Abort the current running prompt in a session.",
responses={404: {"model": ErrorResponse, "description": "Session not found"}},
status_code=200,
)
async def abort_session(sessionID: str) -> Response:
try:
session_manager.abort_session(sessionID)
except KeyError:
raise HTTPException(status_code=404, detail=f"Session not found: {sessionID}")
return Response(status_code=200)
4. Nested path parameter
@app.post(
"/session/{sessionID}/permissions/{permissionID}",
tags=["Session"],
summary="Grant permission",
description="Grant a pending permission request.",
responses={404: {"model": ErrorResponse, "description": "Session not found"}},
status_code=200,
)
async def grant_permission(sessionID: str, permissionID: str) -> Response:
logger.info("Permission granted: session=%s, permission=%s", sessionID, permissionID)
return Response(status_code=200)
Request/Response Models
from pydantic import BaseModel, Field
class CreateSessionRequest(BaseModel):
title: Optional[str] = Field(None, description="Session title")
class PromptPart(BaseModel):
type: str = Field("text", description="Part type: text")
text: str = Field("", description="Text content")
class PromptInput(BaseModel):
parts: List[PromptPart] = Field(default_factory=list, description="Message parts")
agent: Optional[str] = Field(None, description="Agent name to use")
model: Optional[str] = Field(None, description="Model name to use")
Common Imports
from kimix.server.app import create_app
from kimix.server.bus import bus, BusEvent
from kimix.server.session_manager import session_manager
Starting the Server
from kimix.server.app import create_app
import uvicorn
app = create_app()
uvicorn.run(app, host="127.0.0.1", port=4096, log_level="info")
Or via CLI:
kimix serve --host 127.0.0.1 --port 4096
