supermemory/apps/sdk-playground/python/server.py

"""HTTP server for Python SDK chat integrations in the playground."""

import asyncio
import hashlib
import json
import os
import re
import time
from pathlib import Path
from typing import Annotated, Any, Literal, Optional
from urllib.parse import quote, urlparse

from dotenv import load_dotenv
from fastapi import FastAPI, Header, Query
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, SecretStr, model_validator
from starlette.middleware.trustedhost import TrustedHostMiddleware

_root = Path(__file__).resolve().parent
load_dotenv(_root / ".env")
load_dotenv(_root.parent / ".env.local")
load_dotenv(_root.parent / ".env")

DEFAULT_SUPERMEMORY_BASE_URL = "https://api.supermemory.ai"
HTTP_TIMEOUT_SECONDS = 60.0
CHAT_TIMEOUT_SECONDS = 115.0
CONTEXT_DEBUG_TIMEOUT_SECONDS = 10.0
DIRECT_SAVE_TIMEOUT_SECONDS = 10.0
MAX_OUTPUT_TOKENS = 2_048
DOCUMENT_LIST_LIMIT = 25
# Matches the v4 profile search defaults the TS middlewares keep (memories mode, 0.6).
PROFILE_SEARCH_THRESHOLD = 0.6
MAX_MESSAGE_LENGTH = 20_000
MAX_MESSAGES = 64
MAX_TOTAL_MESSAGE_LENGTH = 100_000
MAX_API_KEY_LENGTH = 1_024
MAX_NAMESPACE_LENGTH = 100
MAX_CONVERSATION_ID_LENGTH = 242
NAMESPACE_PATTERN = r"^[a-zA-Z0-9_:-]+$"

TOOLS_SYSTEM_PROMPT = """You are a helpful assistant with Supermemory long-term memory.

You have tools to manage memory. Use them proactively:
- search_memories: hybrid recall — search before answering whenever user-specific context could help (do not wait to be asked)
- get_profile: broad static/dynamic user context at conversation start or when you need a wide overview
- add_memory: store a new generalizable fact
- document_list / document_add / document_delete: manage source documents
- memory_forget: soft-delete one profile fact (not whole documents)

Before answering questions about the user, their preferences, or past context, search memories or get profile first. When the user asks you to remember something, use add_memory."""

app = FastAPI(title="SDK Playground Python Chat")
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["127.0.0.1", "localhost"],
)


class ChatMessage(BaseModel):
    role: Literal["user", "assistant", "system"]
    content: str = Field(max_length=MAX_MESSAGE_LENGTH)


class MiddlewareConfig(BaseModel):
    addMemory: Literal["always", "never"] = "always"
    verbose: bool = False


class PlaygroundInputError(ValueError):
    """A request value is missing after transport-level validation."""


class SupermemoryApiKeys(BaseModel):
    supermemoryApiKey: SecretStr = Field(max_length=MAX_API_KEY_LENGTH)


class ApiKeys(SupermemoryApiKeys):
    openaiApiKey: SecretStr = Field(max_length=MAX_API_KEY_LENGTH)


class ChatRequest(BaseModel):
    sdkId: Literal[
        "py-openai-middleware",
        "py-openai-tools",
        "py-supermemory-direct",
    ]
    messages: list[ChatMessage] = Field(min_length=1, max_length=MAX_MESSAGES)
    namespace: str = Field(
        default="sdk-playground",
        min_length=1,
        max_length=MAX_NAMESPACE_LENGTH,
        pattern=NAMESPACE_PATTERN,
    )
    conversationId: str = Field(
        min_length=1,
        max_length=MAX_CONVERSATION_ID_LENGTH,
    )
    memoryMode: Optional[Literal["profile", "query", "full"]] = "full"
    middlewareConfig: Optional[MiddlewareConfig] = None
    apiKeys: Optional[ApiKeys] = None

    @model_validator(mode="after")
    def require_user_message(self) -> "ChatRequest":
        if not any(
            message.role == "user" and message.content.strip()
            for message in self.messages
        ):
            raise ValueError("messages must include a non-empty user message")
        if (
            sum(len(message.content) for message in self.messages)
            > MAX_TOTAL_MESSAGE_LENGTH
        ):
            raise ValueError(
                f"total message content cannot exceed {MAX_TOTAL_MESSAGE_LENGTH} characters"
            )
        return self


class ContextRequest(BaseModel):
    namespace: str = Field(
        default="sdk-playground",
        min_length=1,
        max_length=MAX_NAMESPACE_LENGTH,
        pattern=NAMESPACE_PATTERN,
    )
    query: Optional[str] = Field(default=None, max_length=MAX_MESSAGE_LENGTH)
    apiKeys: Optional[SupermemoryApiKeys] = None


def model_name() -> str:
    return os.getenv("MODEL_NAME", "gpt-4o-mini")


def supplied_secret(value: Optional[SecretStr], label: str) -> str:
    secret = value.get_secret_value().strip() if value else ""
    if not secret:
        raise PlaygroundInputError(f"{label} must be supplied with the request")
    return secret


def resolve_supermemory_key(api_keys: Optional[SupermemoryApiKeys]) -> str:
    return supplied_secret(
        api_keys.supermemoryApiKey if api_keys else None,
        "Supermemory API key",
    )


def resolve_chat_keys(api_keys: Optional[ApiKeys]) -> tuple[str, str]:
    return (
        resolve_supermemory_key(api_keys),
        supplied_secret(api_keys.openaiApiKey if api_keys else None, "OpenAI API key"),
    )


def supermemory_base_url() -> str:
    configured = os.getenv("SUPERMEMORY_BASE_URL", "").strip()
    base_url = (configured or DEFAULT_SUPERMEMORY_BASE_URL).rstrip("/")
    parsed = urlparse(base_url)
    if parsed.scheme not in ("http", "https") or not parsed.netloc:
        raise RuntimeError("SUPERMEMORY_BASE_URL must be an absolute HTTP(S) URL")
    if parsed.username or parsed.password or parsed.query or parsed.fragment:
        raise RuntimeError(
            "SUPERMEMORY_BASE_URL cannot contain credentials, a query, or a fragment"
        )
    return base_url


def public_error(error: Exception, *secrets: str) -> str:
    message = str(error)
    for secret in secrets:
        if secret:
            message = message.replace(secret, "[redacted]")
    return message[:1_000]


async def chat_openai_middleware(
    messages: list[ChatMessage],
    namespace: str,
    conversation_id: str,
    memory_mode: str,
    middleware_config: MiddlewareConfig,
    sm_key: str,
    oai_key: str,
) -> str:
    from openai import AsyncOpenAI
    from supermemory_openai import OpenAIMiddlewareOptions, with_supermemory

    client = with_supermemory(
        AsyncOpenAI(
            api_key=oai_key,
            timeout=HTTP_TIMEOUT_SECONDS,
            max_retries=1,
        ),
        OpenAIMiddlewareOptions(
            container_tag=namespace,
            custom_id=conversation_id,
            mode=memory_mode,
            add_memory=middleware_config.addMemory,
            verbose=middleware_config.verbose,
            api_key=sm_key,
            base_url=supermemory_base_url(),
        ),
    )

    openai_messages = [m.model_dump() for m in messages]
    if not any(m.role == "system" for m in messages):
        openai_messages.insert(
            0,
            {
                "role": "system",
                "content": (
                    "You are a helpful assistant with long-term memory about the user."
                ),
            },
        )

    response = await client.chat.completions.create(
        model=model_name(),
        messages=openai_messages,
        max_completion_tokens=MAX_OUTPUT_TOKENS,
    )
    return response.choices[0].message.content or ""


async def chat_openai_tools(
    messages: list[ChatMessage],
    namespace: str,
    sm_key: str,
    oai_key: str,
) -> tuple[str, list[dict[str, Any]]]:
    from openai import AsyncOpenAI
    from supermemory_openai import SupermemoryTools, execute_memory_tool_calls

    openai_client = AsyncOpenAI(
        api_key=oai_key,
        timeout=HTTP_TIMEOUT_SECONDS,
        max_retries=1,
    )
    config: dict[str, Any] = {
        "base_url": supermemory_base_url(),
        "container_tags": [namespace],
    }

    tools = SupermemoryTools(sm_key, config)
    tool_defs = tools.get_tool_definitions()
    trace: list[dict[str, Any]] = []

    convo: list[dict[str, Any]] = [
        {"role": "system", "content": TOOLS_SYSTEM_PROMPT},
        *[m.model_dump() for m in messages if m.role != "system"],
    ]

    for step in range(8):
        response = await openai_client.chat.completions.create(
            model=model_name(),
            messages=convo,
            tools=tool_defs,
            max_completion_tokens=MAX_OUTPUT_TOKENS,
        )
        message = response.choices[0].message
        convo.append(message.model_dump())

        if message.tool_calls:
            tool_messages = await execute_memory_tool_calls(
                sm_key,
                message.tool_calls,
                config,
            )
            for i, call in enumerate(message.tool_calls):
                raw = tool_messages[i]["content"]
                try:
                    parsed = json.loads(raw)
                except json.JSONDecodeError:
                    parsed = raw
                trace.append(
                    {
                        "step": step + 1,
                        "toolName": call.function.name,
                        "args": json.loads(call.function.arguments),
                        "result": parsed,
                    }
                )
            convo.extend(tool_messages)
            continue

        return message.content or "", trace

    raise RuntimeError("Tool loop exceeded max steps")


def object_field(value: Any, name: str, default: Any = None) -> Any:
    if isinstance(value, dict):
        return value.get(name, default)
    return getattr(value, name, default)


def list_field(value: Any, name: str) -> list[Any]:
    result = object_field(value, name, [])
    return result if isinstance(result, list) else []


def display_context_item(item: Any) -> str:
    if hasattr(item, "model_dump"):
        return json.dumps(item.model_dump(mode="json"), ensure_ascii=False)
    if isinstance(item, dict):
        return json.dumps(item, ensure_ascii=False)
    return str(item)


def direct_conversation_id(conversation_id: str) -> str:
    readable = re.sub(r"[^A-Za-z0-9._-]+", "-", conversation_id).strip("-._")
    readable = readable[:40] or "session"
    digest = hashlib.sha256(conversation_id.encode("utf-8")).hexdigest()[:12]
    return f"sdk-playground-direct-{readable}-{digest}"


def conversation_transcript(messages: list[ChatMessage], assistant_text: str) -> str:
    transcript = [
        f"{message.role.capitalize()}: {message.content}"
        for message in messages
        if message.role != "system"
    ]
    transcript.append(f"Assistant: {assistant_text or '(empty response)'}")
    return "\n\n".join(transcript)


def supermemory_http(sm_key: str, timeout: float = HTTP_TIMEOUT_SECONDS):
    import httpx

    return httpx.AsyncClient(
        base_url=supermemory_base_url(),
        headers={"Authorization": f"Bearer {sm_key}"},
        timeout=timeout,
        follow_redirects=False,
    )


def namespace_path(namespace: str, suffix: str) -> str:
    return f"/ns/{quote(namespace, safe='')}{suffix}"


async def fetch_profile_context(
    namespace: str,
    sm_key: str,
    query: Optional[str] = None,
) -> dict[str, list[Any]]:
    async with supermemory_http(sm_key) as http:
        profile_request = http.post(namespace_path(namespace, "/profile"), json={})
        search_request = (
            http.post(
                namespace_path(namespace, "/search"),
                json={
                    "query": query,
                    "searchMode": "memories",
                    "threshold": PROFILE_SEARCH_THRESHOLD,
                },
            )
            if query
            else None
        )
        responses = await asyncio.gather(
            profile_request, *([search_request] if search_request else [])
        )
        for response in responses:
            response.raise_for_status()

    profile = responses[0].json().get("profile") or {}
    search = responses[1].json() if len(responses) > 1 else {}
    return {
        "static": profile.get("static") or [],
        "dynamic": profile.get("dynamic") or [],
        "searchResults": search.get("results") or [],
    }


async def chat_supermemory_direct(
    messages: list[ChatMessage],
    namespace: str,
    conversation_id: str,
    sm_key: str,
    oai_key: str,
) -> tuple[str, str, dict[str, list[Any]]]:
    """Manual pattern: profile + search for context, then OpenAI, then add the conversation."""
    from openai import AsyncOpenAI

    openai_client = AsyncOpenAI(
        api_key=oai_key,
        timeout=HTTP_TIMEOUT_SECONDS,
        max_retries=1,
    )

    user_messages = [m for m in messages if m.role == "user"]
    last_user = user_messages[-1].content if user_messages else ""

    profile_context = await fetch_profile_context(namespace, sm_key, last_user)
    context = "\n".join(
        (
            "Profile static: "
            + ", ".join(map(display_context_item, profile_context["static"])),
            "Profile dynamic: "
            + ", ".join(map(display_context_item, profile_context["dynamic"])),
            "Relevant search results: "
            + ", ".join(map(display_context_item, profile_context["searchResults"])),
        )
    )

    openai_messages: list[dict[str, str]] = [
        {
            "role": "system",
            "content": f"You are a helpful assistant. User context:\n{context}",
        },
        *[m.model_dump() for m in messages if m.role != "system"],
    ]

    response = await openai_client.chat.completions.create(
        model=model_name(),
        messages=openai_messages,
        max_completion_tokens=MAX_OUTPUT_TOKENS,
    )
    assistant_text = response.choices[0].message.content or ""

    document_id = direct_conversation_id(conversation_id)
    return assistant_text, document_id, profile_context


async def save_direct_conversation(
    messages: list[ChatMessage],
    assistant_text: str,
    namespace: str,
    document_id: str,
    sm_key: str,
) -> dict[str, Any]:
    try:
        async with asyncio.timeout(DIRECT_SAVE_TIMEOUT_SECONDS):
            async with supermemory_http(sm_key, DIRECT_SAVE_TIMEOUT_SECONDS) as http:
                response = await http.post(
                    namespace_path(namespace, "/document"),
                    json={
                        "content": conversation_transcript(messages, assistant_text),
                        "id": document_id,
                        "dreaming": "instant",
                    },
                )
                response.raise_for_status()
                accepted = response.json()
        return {
            "type": "conversation_save_accepted",
            "label": "Full conversation accepted for processing",
            "detail": {
                "nonFatal": True,
                "namespace": namespace,
                "id": document_id,
                "documentId": accepted.get("id"),
                "status": accepted.get("status"),
            },
        }
    except Exception as error:
        return {
            "type": "conversation_save_failed",
            "label": "Conversation save unavailable",
            "detail": {
                "nonFatal": True,
                "namespace": namespace,
                "id": document_id,
                "error": public_error(error, sm_key),
            },
        }


async def fetch_document_memories(http: Any, namespace: str, document_id: str) -> list[Any]:
    try:
        response = await http.get(
            namespace_path(namespace, f"/document/{quote(document_id, safe='')}"),
            params={"include": "memories"},
        )
        response.raise_for_status()
        return response.json().get("memories") or []
    except Exception:
        return []


async def fetch_namespace_context(
    namespace: str,
    sm_key: str,
    query: Optional[str] = None,
) -> dict[str, Any]:
    if not sm_key:
        raise RuntimeError("Supermemory API key must be supplied")

    profile_context = await fetch_profile_context(namespace, sm_key, query)

    async with supermemory_http(sm_key) as http:
        docs_response = await http.post(
            namespace_path(namespace, "/list/documents"),
            params={
                "limit": DOCUMENT_LIST_LIMIT,
                "sort": "createdAt",
                "order": "desc",
            },
            json={},
        )
        docs_response.raise_for_status()
        docs = docs_response.json()
        raw_documents = docs.get("documents") or []
        # v5 lists omit memories, so each document is fetched with include=memories.
        memories = await asyncio.gather(
            *(fetch_document_memories(http, namespace, doc["id"]) for doc in raw_documents)
        )

    documents = []
    for doc, memory_entries in zip(raw_documents, memories):
        system = doc.get("system") or {}
        documents.append(
            {
                "id": doc.get("id"),
                "title": doc.get("title"),
                "status": system.get("status"),
                "createdAt": system.get("createdAt"),
                "updatedAt": system.get("updatedAt"),
                "summary": doc.get("summary"),
                "memoryEntries": memory_entries,
            }
        )

    return {
        "namespace": namespace,
        "query": query,
        "profile": profile_context,
        "documents": documents,
        "pagination": docs.get("pagination"),
    }


def reconstruct_python_sdk_memory_block(
    memory_mode: str,
    profile: dict[str, Any],
) -> tuple[dict[str, list[str]], str]:
    from supermemory_openai import convert_profile_to_markdown, deduplicate_memories
    from supermemory_openai.utils import wrap_memory_context

    deduplicated = deduplicate_memories(
        static=profile.get("static", []) if memory_mode != "query" else [],
        dynamic=profile.get("dynamic", []) if memory_mode != "query" else [],
        search_results=profile.get("searchResults", []),
    )
    visible_profile = {
        "static": deduplicated.static,
        "dynamic": deduplicated.dynamic,
        "searchResults": (
            [] if memory_mode == "profile" else deduplicated.search_results
        ),
    }

    profile_data = ""
    if memory_mode != "query":
        profile_data = convert_profile_to_markdown(
            {
                "profile": {
                    "static": visible_profile["static"],
                    "dynamic": visible_profile["dynamic"],
                },
                "searchResults": {"results": []},
            }
        )

    search_results_memories = ""
    if memory_mode != "profile" and visible_profile["searchResults"]:
        search_results_memories = (
            "Search results for user's recent message: \n"
            + "\n".join(f"- {memory}" for memory in visible_profile["searchResults"])
        )

    memories = f"{profile_data}\n{search_results_memories}".strip()
    return visible_profile, wrap_memory_context(memories)


def build_middleware_memory_debug(
    namespace: str,
    conversation_id: str,
    memory_mode: str,
    last_user_message: str,
    context: Optional[dict[str, Any]],
    context_error: Optional[str],
    middleware_config: MiddlewareConfig,
) -> list[dict[str, Any]]:
    debug: list[dict[str, Any]] = []
    if context is None:
        debug.append(
            {
                "type": "context_debug_unavailable",
                "label": "Post-response context snapshot unavailable",
                "detail": {"error": context_error or "Unknown context error"},
            }
        )
    else:
        raw_profile = context["profile"]
        profile, memory_block = reconstruct_python_sdk_memory_block(
            memory_mode,
            raw_profile,
        )

        debug.extend(
            (
                {
                    "type": "profile_fetch",
                    "label": "Post-response context reconstruction",
                    "detail": {
                        "authoritativeMiddlewareCapture": False,
                        "timing": "after model response",
                        "endpoint": (
                            "POST /ns/{namespace}/profile + POST /ns/{namespace}/search"
                        ),
                        "namespace": namespace,
                        "id": conversation_id,
                        "memoryMode": memory_mode,
                        "query": context.get("query"),
                        "staticCount": len(profile.get("static", [])),
                        "dynamicCount": len(profile.get("dynamic", [])),
                        "searchResultCount": len(profile.get("searchResults", [])),
                    },
                },
                {
                    "type": "context_preview",
                    "label": (
                        "Reconstructed SDK-owned memory block "
                        "(not middleware capture)"
                    ),
                    "preview": memory_block,
                    "detail": {
                        "totalFacts": (
                            len(profile.get("static", []))
                            + len(profile.get("dynamic", []))
                            + len(profile.get("searchResults", []))
                        ),
                        "fullLength": len(memory_block),
                    },
                },
            )
        )

    save_detail = {
        "namespace": namespace,
        "id": f"conversation:{conversation_id}",
        "addMemory": middleware_config.addMemory,
        "verbose": middleware_config.verbose,
    }
    if middleware_config.addMemory == "always" and last_user_message.strip():
        debug.append(
            {
                "type": "conversation_save_queued",
                "label": "Conversation save queued by middleware",
                "detail": save_detail,
            }
        )
    else:
        debug.append(
            {
                "type": "conversation_save_skipped",
                "label": "Conversation save disabled",
                "detail": save_detail,
            }
        )
    return debug


async def fetch_context_for_debug(
    namespace: str,
    query: Optional[str],
    sm_key: str,
) -> tuple[Optional[dict[str, Any]], Optional[str]]:
    try:
        async with asyncio.timeout(CONTEXT_DEBUG_TIMEOUT_SECONDS):
            profile = await fetch_profile_context(namespace, sm_key, query)
            return (
                {
                    "namespace": namespace,
                    "query": query,
                    "profile": profile,
                },
                None,
            )
    except Exception as error:
        return None, public_error(error, sm_key)


@app.get("/context")
async def context_get(
    namespace: Annotated[
        str,
        Query(
            min_length=1,
            max_length=MAX_NAMESPACE_LENGTH,
            pattern=NAMESPACE_PATTERN,
        ),
    ] = "sdk-playground",
    query: Annotated[Optional[str], Query(max_length=MAX_MESSAGE_LENGTH)] = None,
    x_supermemory_api_key: Annotated[
        Optional[str],
        Header(alias="X-Supermemory-API-Key"),
    ] = None,
):
    sm_key = ""
    try:
        sm_key = supplied_secret(
            SecretStr(x_supermemory_api_key) if x_supermemory_api_key else None,
            "X-Supermemory-API-Key header",
        )
        async with asyncio.timeout(HTTP_TIMEOUT_SECONDS):
            ctx = await fetch_namespace_context(namespace, sm_key, query)
        return {"ok": True, "context": ctx}
    except Exception as error:
        return JSONResponse(
            status_code=(
                504
                if isinstance(error, TimeoutError)
                else 400 if isinstance(error, PlaygroundInputError) else 500
            ),
            content={"ok": False, "error": public_error(error, sm_key)},
        )


@app.post("/context")
async def context_post(req: ContextRequest):
    sm_key = ""
    try:
        sm_key = resolve_supermemory_key(req.apiKeys)
        async with asyncio.timeout(HTTP_TIMEOUT_SECONDS):
            ctx = await fetch_namespace_context(req.namespace, sm_key, req.query)
        return {"ok": True, "context": ctx}
    except Exception as error:
        return JSONResponse(
            status_code=(
                504
                if isinstance(error, TimeoutError)
                else 400 if isinstance(error, PlaygroundInputError) else 500
            ),
            content={"ok": False, "error": public_error(error, sm_key)},
        )


@app.get("/health")
async def health():
    return {
        "ok": True,
        "playground": "sdk-playground",
        "requiresRequestKeys": True,
        "model": model_name(),
        "sdks": [
            "py-openai-middleware",
            "py-openai-tools",
            "py-supermemory-direct",
        ],
    }


@app.post("/chat")
async def chat(req: ChatRequest):
    started = time.time()
    sm_key = ""
    oai_key = ""
    try:
        sm_key, oai_key = resolve_chat_keys(req.apiKeys)
        tool_trace: list[dict[str, Any]] = []
        memory_debug: list[dict[str, Any]] = []
        middleware_debug: Optional[tuple[MiddlewareConfig, str, Optional[str]]] = None
        direct_debug: Optional[tuple[str, dict[str, list[Any]], str]] = None
        async with asyncio.timeout(CHAT_TIMEOUT_SECONDS):
            if req.sdkId == "py-openai-middleware":
                middleware_config = req.middlewareConfig or MiddlewareConfig()
                text = await chat_openai_middleware(
                    req.messages,
                    req.namespace,
                    req.conversationId,
                    req.memoryMode or "full",
                    middleware_config,
                    sm_key,
                    oai_key,
                )
                last_user = next(
                    (m.content for m in reversed(req.messages) if m.role == "user"),
                    "",
                )
                query = last_user if req.memoryMode != "profile" else None
                middleware_debug = (middleware_config, last_user, query)
            elif req.sdkId == "py-openai-tools":
                text, tool_trace = await chat_openai_tools(
                    req.messages, req.namespace, sm_key, oai_key
                )
            elif req.sdkId == "py-supermemory-direct":
                text, document_id, profile_context = await chat_supermemory_direct(
                    req.messages,
                    req.namespace,
                    req.conversationId,
                    sm_key,
                    oai_key,
                )
                last_user = next(
                    (m.content for m in reversed(req.messages) if m.role == "user"),
                    "",
                )
                direct_debug = (document_id, profile_context, last_user)
            else:
                raise RuntimeError(f"Unsupported Python SDK: {req.sdkId}")

        if middleware_debug is not None:
            middleware_config, last_user, query = middleware_debug
            ctx, context_error = await fetch_context_for_debug(
                req.namespace,
                query,
                sm_key,
            )
            memory_debug = build_middleware_memory_debug(
                req.namespace,
                req.conversationId,
                req.memoryMode or "full",
                last_user,
                ctx,
                context_error,
                middleware_config,
            )
        elif direct_debug is not None:
            document_id, profile_context, last_user = direct_debug
            save_debug = await save_direct_conversation(
                req.messages,
                text,
                req.namespace,
                document_id,
                sm_key,
            )
            memory_debug = [
                {
                    "type": "manual_profile",
                    "label": "Profile context used for this response",
                    "detail": {
                        "namespace": req.namespace,
                        "query": last_user,
                        "staticCount": len(profile_context["static"]),
                        "dynamicCount": len(profile_context["dynamic"]),
                        "searchResultCount": len(profile_context["searchResults"]),
                    },
                },
                save_debug,
            ]

        return {
            "ok": True,
            "sdkId": req.sdkId,
            "message": {"role": "assistant", "content": text},
            "toolTrace": tool_trace,
            "memoryDebug": memory_debug,
            "durationMs": int((time.time() - started) * 1000),
        }
    except TimeoutError:
        return JSONResponse(
            status_code=504,
            content={
                "ok": False,
                "sdkId": req.sdkId,
                "error": f"Python chat timed out after {int(CHAT_TIMEOUT_SECONDS)} seconds",
                "durationMs": int((time.time() - started) * 1000),
            },
        )
    except Exception as error:
        return JSONResponse(
            status_code=400 if isinstance(error, PlaygroundInputError) else 500,
            content={
                "ok": False,
                "sdkId": req.sdkId,
                "error": public_error(error, sm_key, oai_key),
                "durationMs": int((time.time() - started) * 1000),
            },
        )


if __name__ == "__main__":
    import uvicorn

    port = int(os.getenv("SDK_PLAYGROUND_PYTHON_PORT", "8792"))
    uvicorn.run(app, host="127.0.0.1", port=port, log_level="info")