Frankenstein-Labs/Cortex-ai

1"""OpenAI-compatible HTTP API for CORTEX AI.2 3Any client that speaks the OpenAI chat-completions protocol -- the official4Python SDK, LangChain, LlamaIndex, a curl script -- can talk to CORTEX AI by5changing only the base URL.6 7Endpoints:8    GET  /health9    GET  /v1/models10    POST /v1/chat/completions11"""12 13from __future__ import annotations14 15import time16import uuid17from typing import Any18 19from fastapi import FastAPI, Header, HTTPException20from pydantic import BaseModel, Field21 22from ..adapters.base import ModelAdapter23from ..config import CortexConfig24from ..engine.agent import CortexAgent25from ..identity import identity_language, identity_response, is_identity_question26from ..tools.registry import registry_from_names27 28 29class ChatMessage(BaseModel):30    role: str31    content: str = ""32    name: str | None = None33 34 35class ChatCompletionRequest(BaseModel):36    model: str | None = None37    messages: list[ChatMessage]38    temperature: float | None = None39    max_tokens: int | None = None40    stream: bool = False41    tools: list[dict[str, Any]] | None = None42 43 44class Usage(BaseModel):45    prompt_tokens: int = 046    completion_tokens: int = 047    total_tokens: int = 048 49 50class ChatCompletionChoice(BaseModel):51    index: int = 052    message: ChatMessage53    finish_reason: str = "stop"54 55 56class ChatCompletionResponse(BaseModel):57    id: str58    object: str = "chat.completion"59    created: int60    model: str61    choices: list[ChatCompletionChoice]62    usage: Usage63    # CORTEX extension: the reasoning trace, when thinking mode is on.64    reasoning_content: str = ""65    tool_calls: list[dict[str, Any]] = Field(default_factory=list)66 67 68def create_app(adapter: ModelAdapter, config: CortexConfig | None = None) -> FastAPI:69    """Build the FastAPI application around a model adapter."""70    cfg = config or CortexConfig()71    tools = registry_from_names(cfg.enabled_tools)72    agent = CortexAgent(73        adapter,74        tools,75        cfg.engine,76        system_prompt=cfg.system_prompt,77    )78 79    app = FastAPI(80        title="CORTEX AI API",81        version="1.0.0",82        description="API compatible OpenAI pour CORTEX AI, un projet de Frankenstein-Labs.",83    )84    app.state.cortex_config = cfg85    app.state.cortex_agent = agent86    app.state.identity_interception = True87 88    def _check_auth(authorization: str | None) -> None:89        if not cfg.server.requires_auth:90            return91        expected = f"Bearer {cfg.server.api_key}"92        if authorization != expected:93            raise HTTPException(status_code=401, detail="invalid API key")94 95    @app.get("/health")96    def health() -> dict[str, Any]:97        return {98            "status": "ok",99            "model": cfg.model_id,100            "tools": tools.names(),101            "thinking_mode": cfg.engine.thinking_mode,102            "identity_interception": "deterministic",103        }104 105    @app.get("/v1/models")106    def list_models(authorization: str | None = Header(default=None)) -> dict[str, Any]:107        _check_auth(authorization)108        return {109            "object": "list",110            "data": [111                {112                    "id": cfg.model_id,113                    "object": "model",114                    "created": int(time.time()),115                    "owned_by": "Frankenstein-Labs",116                }117            ],118        }119 120    @app.post("/v1/chat/completions", response_model=ChatCompletionResponse)121    def chat_completions(122        request: ChatCompletionRequest,123        authorization: str | None = Header(default=None),124    ) -> ChatCompletionResponse:125        _check_auth(authorization)126 127        if request.stream:128            raise HTTPException(129                status_code=400,130                detail="stream=true is not supported yet; use stream=false",131            )132        if not request.messages:133            raise HTTPException(status_code=400, detail="messages must not be empty")134 135        # Deterministic identity boundary: answer before system prompts, tools,136        # or model inference can alter the canonical creator attribution.137        last_user_message = next(138            (m.content for m in reversed(request.messages) if m.role == "user"),139            None,140        )141        if last_user_message is not None and is_identity_question(last_user_message):142            content = identity_response(identity_language(last_user_message))143            prompt_tokens = adapter.count_tokens(last_user_message)144            completion_tokens = adapter.count_tokens(content)145            return ChatCompletionResponse(146                id=f"chatcmpl-{uuid.uuid4().hex[:24]}",147                created=int(time.time()),148                model=request.model or cfg.model_id,149                choices=[150                    ChatCompletionChoice(151                        index=0,152                        message=ChatMessage(role="assistant", content=content),153                        finish_reason="stop",154                    )155                ],156                usage=Usage(157                    prompt_tokens=prompt_tokens,158                    completion_tokens=completion_tokens,159                    total_tokens=prompt_tokens + completion_tokens,160                ),161            )162 163        system_override: list[dict[str, Any]] = []164        turns: list[dict[str, Any]] = []165        for m in request.messages:166            if m.role == "system":167                system_override.append({"role": "system", "content": m.content})168            else:169                turns.append({"role": m.role, "content": m.content})170 171        if system_override:172            agent.system_prompt = system_override[-1]["content"]173 174        result = agent.run(turns)175        prompt_tokens = sum(adapter.count_tokens(m["content"]) for m in turns)176        completion_tokens = adapter.count_tokens(result.content)177 178        return ChatCompletionResponse(179            id=f"chatcmpl-{uuid.uuid4().hex[:24]}",180            created=int(time.time()),181            model=request.model or cfg.model_id,182            choices=[183                ChatCompletionChoice(184                    index=0,185                    message=ChatMessage(role="assistant", content=result.content),186                    finish_reason="stop",187                )188            ],189            usage=Usage(190                prompt_tokens=prompt_tokens,191                completion_tokens=completion_tokens,192                total_tokens=prompt_tokens + completion_tokens,193            ),194            reasoning_content=result.reasoning,195            tool_calls=[196                {"name": c.name, "arguments": c.arguments, "result": c.result, "ok": c.ok}197                for c in result.tool_calls198            ],199        )200 201    return app202