From 1cb590db2323f568f56a4253d14c4a83a407ee02 Mon Sep 17 00:00:00 2001 From: Speedliner Date: Thu, 23 Jul 2026 17:44:59 +0200 Subject: [PATCH 01/10] Update claude_agent_pipe.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add OpenWebUI Tools/MCP passthrough, fix multi-arity bug, bump to 0.2 - Fix _build_kb_mcp_server: the no-knowledge branch returned a 2-tuple (None, []) while every call site unpacked 3 values, raising ValueError: not enough values to unpack (expected 3, got 2) on any turn without an attached knowledge base. Now consistently returns (None, [], {}). - Add __tools__ passthrough: wrap OpenWebUI's Tools / external tool servers (incl. MCP via mcpo) attached to the Workspace Model as an in-process MCP server ("owui-tools"), merged alongside the existing "helm-kb" server. Lets Claude Code use whatever tools/connectors are configured in OpenWebUI natively, without hardcoding a server URL in the pipe — stays in sync if the attached tools change later. New: _JSON_SCHEMA_TYPE_MAP, _build_owui_tools_mcp_server(). Known limitation: OpenWebUI's built-in tools (web_search, image_generation, execute_code) aren't included in __tools__ yet (upstream limitation); only user-defined Tools and external/MCP tool servers are. Not an issue here since Claude Code already ships its own WebSearch/WebFetch. - version: 0.1 -> 0.2, contributors: Speedliner (author unchanged) --- claude_agent_pipe.py | 101 ++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 94 insertions(+), 7 deletions(-) diff --git a/claude_agent_pipe.py b/claude_agent_pipe.py index 8da0ed1..0b8d6b8 100644 --- a/claude_agent_pipe.py +++ b/claude_agent_pipe.py @@ -2,7 +2,8 @@ title: Claude Code description: Run Claude Code's agent loop from inside OpenWebUI chats via the Claude Agent SDK. author: Thomas Friedel -version: 0.1 +contributors: Speedliner +version: 0.2 license: MIT requirements: claude-agent-sdk>=0.1.60, anthropic>=0.40.0 """ @@ -379,7 +380,7 @@ def _build_kb_mcp_server( that OpenWebUI's middleware already filtered by the user's grants. """ if not knowledge: - return None, [] + return None, [], {} collection_names = [k["id"] for k in knowledge] display = ", ".join(k["name"] for k in knowledge) @@ -737,6 +738,82 @@ def _build_kb_mcp_server( return server, tool_names, tools_by_name +# --------------------------------------------------------------------------- +# OpenWebUI Tools/MCP passthrough — wraps whatever Tools or external tool +# servers (incl. MCP, via mcpo) are attached to the Workspace Model as an +# in-process MCP server, so Claude Code gets the same toolbox OpenWebUI's +# native tool-calling would use. Whatever's attached in +# Workspace -> Models -> Tools shows up here automatically at request time — +# no hardcoded server URL/config needed, so it stays in sync if the +# attached tools/connections change later. +# --------------------------------------------------------------------------- + +_JSON_SCHEMA_TYPE_MAP = { + "string": str, + "integer": int, + "number": float, + "boolean": bool, + "array": list, + "object": dict, +} + + +def _build_owui_tools_mcp_server(tools: Optional[Dict[str, Any]]): + """Wrap OpenWebUI's __tools__ (Tools + external/MCP tool servers attached + to the Workspace Model) as an in-process MCP server. Each entry already + carries an OpenAI-style JSON-schema `spec` and an async-wrapped + `callable` — OpenWebUI has already injected any special + __user__/__event_emitter__ params via apply_extra_params_to_tool_function, + so we just need to call it. + + Note: OpenWebUI's built-in tools (web_search, image_generation, + execute_code) are currently NOT included in __tools__ — only + user-defined Tools and external/MCP tool servers are. Claude Code + already ships its own WebSearch/WebFetch tools, so this mainly matters + for custom/internal tool servers. + """ + if not tools: + return None, [] + + sdk_tools = [] + tool_names: List[str] = [] + for name, entry in tools.items(): + spec = entry.get("spec") or {} + params = spec.get("parameters", {}).get("properties", {}) or {} + # Flat type mapping — covers the vast majority of OpenWebUI tool + # specs. Nested/array-item schemas aren't modeled; extend here if a + # specific tool needs richer typing. + input_schema = { + pname: _JSON_SCHEMA_TYPE_MAP.get(pinfo.get("type"), str) + for pname, pinfo in params.items() + } + callable_fn = entry.get("callable") + description = spec.get("description") or f"OpenWebUI tool: {name}" + + def _make_handler(fn: Callable) -> Callable: + async def _handler(args: Dict[str, Any]) -> Dict[str, Any]: + try: + result = fn(**(args or {})) + if asyncio.iscoroutine(result): + result = await result + except Exception as exc: + log.exception("OpenWebUI tool failed") + return { + "content": [{"type": "text", "text": f"Tool failed: {exc}"}] + } + return {"content": [{"type": "text", "text": str(result)}]} + + return _handler + + sdk_tools.append( + tool(name, description, input_schema)(_make_handler(callable_fn)) + ) + tool_names.append(f"mcp__owui-tools__{name}") + + server = create_sdk_mcp_server("owui-tools", "0.1", tools=sdk_tools) + return server, tool_names + + def _anthropic_kb_tool_defs( knowledge: List[Dict[str, str]], has_kb_ids: bool ) -> List[Dict[str, Any]]: @@ -1124,9 +1201,7 @@ class Pipe: final = await stream.get_final_message() except Exception as exc: log.exception("Fast path failed") - yield ( - f"\n\n**Fast-path error:** `{type(exc).__name__}: {exc}`\n" - ) + yield (f"\n\n**Fast-path error:** `{type(exc).__name__}: {exc}`\n") return if final.stop_reason != "tool_use": @@ -1370,6 +1445,7 @@ class Pipe: __files__: Optional[List[Dict[str, Any]]] = None, __user__: Optional[Dict[str, Any]] = None, __metadata__: Optional[Dict[str, Any]] = None, + __tools__: Optional[Dict[str, Any]] = None, ) -> AsyncGenerator[str, None]: # Auth selection: # 1. If CLAUDE_CODE_OAUTH_TOKEN valve is set → use subscription. @@ -1415,7 +1491,12 @@ class Pipe: user_dict=__user__, event_emitter=__event_emitter__, ) - allowed_tools = allowed_tools + kb_tool_names + + # OpenWebUI Tools / external tool servers (incl. MCP via mcpo) + # attached to the Workspace Model → same passthrough treatment. + owui_server, owui_tool_names = _build_owui_tools_mcp_server(__tools__) + + allowed_tools = allowed_tools + kb_tool_names + owui_tool_names options_kwargs: Dict[str, Any] = { "cwd": str(workdir), @@ -1435,8 +1516,14 @@ class Pipe: options_kwargs["resume"] = resume_id if self.valves.MAX_TURNS: options_kwargs["max_turns"] = self.valves.MAX_TURNS + + mcp_servers: Dict[str, Any] = {} if kb_server is not None: - options_kwargs["mcp_servers"] = {"helm-kb": kb_server} + mcp_servers["helm-kb"] = kb_server + if owui_server is not None: + mcp_servers["owui-tools"] = owui_server + if mcp_servers: + options_kwargs["mcp_servers"] = mcp_servers # Extend Claude Code's default agent-loop system prompt with whatever # the Workspace Model configured. `append` keeps the agentic prompt From ae4955834dd7834e86c1af068b5ec7c07941fe51 Mon Sep 17 00:00:00 2001 From: Speedliner Date: Thu, 23 Jul 2026 17:56:56 +0200 Subject: [PATCH 02/10] Update README.md MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit docs: document setup-token scope bug and Docker credential persistence `claude setup-token` has been reported to sometimes issue OAuth tokens scoped too narrowly for API use (anthropics/claude-code#23703), resulting in 401 "Invalid bearer token" on every pipe request despite a seemingly valid token. Ran into this directly; switching to a full interactive `claude` login resolved it immediately. That in turn surfaces an undocumented Docker gotcha: interactive-login credentials are written to ~/.claude and ~/.claude.json inside the container, which are lost on container recreation unless explicitly volume-mounted (the CLAUDE_CODE_OAUTH_TOKEN Valve itself is fine, since Valves persist in OpenWebUI's DB — it's the filesystem-based login that isn't). Added a Docker / persistence notes section with the volume config and one-time login command needed to make this durable. --- README.md | 102 ++++++++++++++++++++++-------------------------------- 1 file changed, 41 insertions(+), 61 deletions(-) diff --git a/README.md b/README.md index fc6dfd8..53160a5 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ This is an Open WebUI **Pipe** that exposes Claude Code as a selectable model. E - **Per-chat workspaces** — each `chat_id` gets a sandboxed working directory that persists across turns - **Dual auth** — bring your own Anthropic **API key** (pay-per-token) *or* a **Claude Pro/Max OAuth token** (bills against your subscription) - **Streaming UI** — tool calls render inline with previews; generated images/PDFs/CSVs surface as artifacts in the chat -- **Configurable valves** — model, permission mode, tool allowlist, max turns, workspace root, setting sources (`CLAUDE.md`) +- **Configurable valves** — model, permission mode, tool allowlist, max turns, workspace root ## Requirements @@ -23,73 +23,24 @@ This is an Open WebUI **Pipe** that exposes Claude Code as a selectable model. E ## Installation 1. In Open WebUI, go to **Workspace → Functions → +** (or **Admin Panel → Functions**). -2. Paste the contents of [`claude_agent_pipe.py`](./claude_agent_pipe.py) into the editor. +2. Paste the contents of [`claude_agent_pipe.py`](https://github.com/tfriedel/openwebui-claude-code/blob/main/claude_agent_pipe.py) into the editor. 3. Save and enable the function. 4. Open the function's **Valves** and configure auth (one of): - `ANTHROPIC_API_KEY` — standard pay-per-token billing - - `CLAUDE_CODE_OAUTH_TOKEN` — generate on a machine with a browser via `claude setup-token`; bills against your Pro/Max/Team subscription + - `CLAUDE_CODE_OAUTH_TOKEN` — generate via `claude setup-token`; bills against your Pro/Max/Team subscription. **Known issue:** `setup-token` has occasionally issued tokens with a restricted scope (see [anthropics/claude-code#23703](https://github.com/anthropics/claude-code/issues/23703)), causing `401 Invalid bearer token` on every request even though the token looks valid. If this happens, log in interactively instead (run `claude` with no arguments, complete the full browser OAuth flow) and persist the resulting credentials — see [Docker / persistence notes](#docker--persistence-notes) below. 5. A new model named **Claude Code** will appear in the model picker. ## Configuration (Valves) -| Valve | Default | Description | -| --- | --- | --- | -| `ANTHROPIC_API_KEY` | *(env)* | Anthropic API key. Falls back to the backend's env var. | -| `CLAUDE_CODE_OAUTH_TOKEN` | *(empty)* | Claude subscription OAuth token. Takes priority over the API key when set. | -| `MODEL` | `claude-haiku-4-5` | Claude model ID (e.g. `claude-haiku-4-5`, `claude-sonnet-4-6`, `claude-opus-4-7`). | -| `PERMISSION_MODE` | `bypassPermissions` | `default`, `acceptEdits`, `bypassPermissions`, `plan`, or `dontAsk`. | -| `ALLOWED_TOOLS` | `Read,Write,Edit,Bash,Glob,Grep,WebSearch,WebFetch` | Comma-separated tools auto-approved without prompting. | -| `WORKDIR_ROOT` | `/tmp/claude-agent-pipe` | Root directory for per-chat workspaces. | -| `MAX_TURNS` | `30` | Max agent turns per user message. `0` disables the cap. | -| `SETTING_SOURCES` | *(empty)* | Comma-separated filesystem setting sources to load: `user`, `project`, `local`. Empty = none (isolated baseline). See below. | - -## Persistent context via `CLAUDE.md` (`SETTING_SOURCES`) - -By default the pipe passes `setting_sources=[]` to the SDK, so **no** filesystem -settings are loaded: each chat starts from a clean baseline and does **not** -inherit the backend user's `~/.claude/` or the workdir's `.claude/`. This is the -safe default for shared deployments. - -If you run a single-user/homelab instance and want persistent environmental -context (e.g. a host inventory or standing instructions in -`~/.claude/CLAUDE.md`) without re-explaining it every chat, set the valve: - -| Value | Loads | -| --- | --- | -| *(empty)* | Nothing — isolated baseline (default). | -| `user` | `~/.claude/CLAUDE.md` **and** `~/.claude/settings.json`. | -| `user,project,local` | Above plus the workdir's `.claude/settings.json` and `.claude/settings.local.json`. | - -Each token maps to one source — `user` → `~/.claude/`, `project` → -`/.claude/settings.json`, `local` → `/.claude/settings.local.json`. -Unknown tokens are dropped. - -> [!WARNING] -> **Settings sources load more than `CLAUDE.md`.** A loaded `settings.json` can -> define **hooks that execute shell commands**, permission grants, env vars, and -> MCP servers — for *every chat*, under the backend user's identity, with the -> pipe's default `bypassPermissions` mode. Only enable `SETTING_SOURCES` on an -> instance you fully trust and control. **Do not enable it on multi-user or -> public deployments** — it breaks per-chat isolation and lets host config -> influence (or run code in) every user's session. There is no way to load -> `CLAUDE.md` *without* also loading `settings.json` from the same source; that -> coupling is in Claude Code, not this pipe. - -### Does this apply to the sandboxed pipe? - -Not directly. [`claude_agent_pipe_sandboxed.py`](./claude_agent_pipe_sandboxed.py) -doesn't use `setting_sources` at all — it shells the `claude` CLI inside an -open-terminal sandbox with a **per-chat `CLAUDE_CONFIG_DIR`** that's created -fresh each chat, so there's no host `~/.claude/` to inherit and nothing to -disable. Persistent context there is meant to come from mechanisms already built -for isolation: - -- **Workspace Model system prompt** — appended on every turn (`--append-system-prompt`); the natural place for standing instructions. -- **Baking into the image** — skills are already vendored into the sandbox image at build time; a `CLAUDE.md` or settings can be baked the same way (under the per-chat `CLAUDE_CONFIG_DIR` layout) if you want file-based context. - -Because the sandbox isolates the agent and a proxy holds the credentials, the -security tradeoff is far milder there — but the `setting_sources` valve itself -has nothing to act on, so it's intentionally **not** added to the sandboxed pipe. +| Valve | Default | Description | +| ------------------------- | ---------------------------------------------------- | ----------------------------------------------------------------------------------- | +| `ANTHROPIC_API_KEY` | *(env)* | Anthropic API key. Falls back to the backend's env var. | +| `CLAUDE_CODE_OAUTH_TOKEN` | *(empty)* | Claude subscription OAuth token. Takes priority over the API key when set. | +| `MODEL` | `claude-haiku-4-5` | Claude model ID (e.g. `claude-haiku-4-5`, `claude-sonnet-4-6`, `claude-opus-4-7`). | +| `PERMISSION_MODE` | `bypassPermissions` | `default`, `acceptEdits`, `bypassPermissions`, `plan`, or `dontAsk`. | +| `ALLOWED_TOOLS` | `Read,Write,Edit,Bash,Glob,Grep,WebSearch,WebFetch` | Comma-separated tools auto-approved without prompting. | +| `WORKDIR_ROOT` | `/tmp/claude-agent-pipe` | Root directory for per-chat workspaces. | +| `MAX_TURNS` | `30` | Max agent turns per user message. `0` disables the cap. | ## Auth notes @@ -97,6 +48,35 @@ When both auth methods are present, the OAuth token wins and the API key is unse Per Anthropic's terms: a Claude subscription is for personal use — **don't re-offer subscription auth to other end users** through a shared Open WebUI deployment. For multi-user setups, use API keys. +## Docker / persistence notes + +If Open WebUI's backend runs in a container without a volume for `~/.claude` (and `~/.claude.json`), credentials from an interactive `claude` login are lost the next time the container is recreated (rebuild, `docker compose down`, etc.) — even though the `CLAUDE_CODE_OAUTH_TOKEN` **Valve** itself survives fine, since Valves live in Open WebUI's own database, not the container filesystem. + +To make an interactive login durable, mount both paths: + +```yaml +services: + openwebui: + volumes: + - open-webui:/app/backend/data + - claude-code-config:/root/.claude + - ./claude-code-auth/.claude.json:/root/.claude.json # pre-create as an empty file on the host + +volumes: + open-webui: + claude-code-config: +``` + +Then log in once inside the running container: + +```bash +docker exec -it /usr/local/lib/python3.11/site-packages/claude_agent_sdk/_bundled/claude +``` + +(path depends on your Python/SDK install — check with `pip show claude-agent-sdk` if it's elsewhere) + +With credentials persisted this way, leave the `CLAUDE_CODE_OAUTH_TOKEN` valve empty — the pipe falls back to whatever the backend environment / `~/.claude` already provides. + ## License MIT From ee6fc4c3c5f76cd94a4123c78b9c9a79d5be3dab Mon Sep 17 00:00:00 2001 From: Speedliner Date: Fri, 7 Aug 2026 15:42:09 +0200 Subject: [PATCH 03/10] Update claude_agent_pipe.py --- claude_agent_pipe.py | 58 ++++++++++++++++++++++++++++++++++---------- 1 file changed, 45 insertions(+), 13 deletions(-) diff --git a/claude_agent_pipe.py b/claude_agent_pipe.py index 0b8d6b8..bf5d509 100644 --- a/claude_agent_pipe.py +++ b/claude_agent_pipe.py @@ -44,19 +44,20 @@ _ARTIFACT_EXTENSIONS = _IMAGE_EXTENSIONS | _DOWNLOAD_EXTENSIONS # when large — this is only a "don't accidentally ship a DVD ISO" guard. _MAX_ARTIFACT_BYTES = 50 * 1024 * 1024 # 50 MiB -from claude_agent_sdk import ( - AssistantMessage, - ClaudeAgentOptions, - ClaudeSDKClient, - ResultMessage, - StreamEvent, - SystemMessage, - ToolResultBlock, - ToolUseBlock, - UserMessage, - create_sdk_mcp_server, - tool, -) + from claude_agent_sdk import ( + AssistantMessage, + ClaudeAgentOptions, + ClaudeSDKClient, + RateLimitEvent, + ResultMessage, + StreamEvent, + SystemMessage, + ToolResultBlock, + ToolUseBlock, + UserMessage, + create_sdk_mcp_server, + tool, + ) log = logging.getLogger(__name__) @@ -77,6 +78,26 @@ _TOOL_PREVIEW_FIELDS = { "Task": "description", } +def _format_rate_limit(info) -> str: + """Human-readable line for a RateLimitEvent.rate_limit_info.""" + import datetime + + parts = [] + if info.status == "rejected": + parts.append("🛑 API-Limit erreicht") + elif info.status == "allowed_warning": + parts.append("⚠️ API-Limit fast erreicht") + else: + parts.append(f"ℹ️ Rate-Limit-Status: {info.status}") + if info.rate_limit_type: + parts.append(f"({info.rate_limit_type})") + if info.utilization is not None: + parts.append(f"· {info.utilization * 100:.0f}% genutzt") + if info.resets_at: + reset_str = datetime.datetime.fromtimestamp(info.resets_at).strftime("%H:%M:%S") + parts.append(f"· verfügbar wieder ab {reset_str}") + return " ".join(parts) + def _tool_preview(name: str, tool_input: Dict[str, Any]) -> str: key = _TOOL_PREVIEW_FIELDS.get(name) @@ -1696,6 +1717,15 @@ class Pipe: ) continue + if isinstance(message, RateLimitEvent): + info = message.rate_limit_info + line = _format_rate_limit(info) + await emit_status(line, done=(info.status == "rejected")) + if info.status in ("allowed_warning", "rejected"): + yield f"\n\n_{line}_\n" + continue + + if isinstance(message, ResultMessage): await emit_status("Done.", done=True) for chunk in _inline_new_artifacts( @@ -1706,6 +1736,8 @@ class Pipe: yield chunk if message.subtype != "success": yield f"\n\n_Agent stopped: {message.subtype}_\n" + if message.api_error_status: + yield f"\n\n**API-Fehler:** `{message.api_error_status}`\n" if message.total_cost_usd is not None: yield f"\n\n_Cost: ${message.total_cost_usd:.4f} · {message.duration_ms}ms_\n" return From 6ac8d3b52cab99de21353ee76cb94dee1f205647 Mon Sep 17 00:00:00 2001 From: Speedliner Date: Fri, 7 Aug 2026 15:44:42 +0200 Subject: [PATCH 04/10] Update claude_agent_pipe_sandboxed.py --- claude_agent_pipe_sandboxed.py | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/claude_agent_pipe_sandboxed.py b/claude_agent_pipe_sandboxed.py index 93bb850..94605b4 100644 --- a/claude_agent_pipe_sandboxed.py +++ b/claude_agent_pipe_sandboxed.py @@ -994,10 +994,40 @@ async def _handle_event( ) return + if etype == "rate_limit_event": + info = event.get("rate_limit_info") or {} + status = info.get("status") + line_parts = [] + if status == "rejected": + line_parts.append("🛑 API-Limit erreicht") + elif status == "allowed_warning": + line_parts.append("⚠️ API-Limit fast erreicht") + else: + line_parts.append(f"ℹ️ Rate-Limit-Status: {status}") + rl_type = info.get("rateLimitType") + if rl_type: + line_parts.append(f"({rl_type})") + utilization = info.get("utilization") + if utilization is not None: + line_parts.append(f"· {utilization * 100:.0f}% genutzt") + resets_at = info.get("resetsAt") + if resets_at: + import datetime + reset_str = datetime.datetime.fromtimestamp(resets_at).strftime("%H:%M:%S") + line_parts.append(f"· verfügbar wieder ab {reset_str}") + line = " ".join(line_parts) + await emit_status(line, done=(status == "rejected")) + if status in ("allowed_warning", "rejected"): + yield f"\n\n_{line}_\n" + return + if etype == "result": subtype = event.get("subtype") if subtype and subtype != "success": yield f"\n\n_Agent stopped: {subtype}_\n" + api_error_status = event.get("api_error_status") + if api_error_status: + yield f"\n\n**API-Fehler:** `{api_error_status}`\n" cost = event.get("total_cost_usd") dur = event.get("duration_ms") if cost is not None and dur is not None: From 4387685614ee6b40e8eb86f6a657b6f07f44bef9 Mon Sep 17 00:00:00 2001 From: Speedliner Date: Fri, 7 Aug 2026 15:45:09 +0200 Subject: [PATCH 05/10] Update claude_agent_pipe.py --- claude_agent_pipe.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/claude_agent_pipe.py b/claude_agent_pipe.py index bf5d509..7fc4019 100644 --- a/claude_agent_pipe.py +++ b/claude_agent_pipe.py @@ -3,7 +3,7 @@ title: Claude Code description: Run Claude Code's agent loop from inside OpenWebUI chats via the Claude Agent SDK. author: Thomas Friedel contributors: Speedliner -version: 0.2 +version: 0.3 license: MIT requirements: claude-agent-sdk>=0.1.60, anthropic>=0.40.0 """ From 3b3092dc75fa399d3f13e62a0657685cddba0d67 Mon Sep 17 00:00:00 2001 From: Speedliner Date: Fri, 7 Aug 2026 15:45:22 +0200 Subject: [PATCH 06/10] Update claude_agent_pipe_sandboxed.py --- claude_agent_pipe_sandboxed.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/claude_agent_pipe_sandboxed.py b/claude_agent_pipe_sandboxed.py index 94605b4..3c77df7 100644 --- a/claude_agent_pipe_sandboxed.py +++ b/claude_agent_pipe_sandboxed.py @@ -4,7 +4,8 @@ description: Run Claude Code inside an open-webui/open-terminal sandbox, one Linux account per OpenWebUI user. Isolates the agent's file/process reach from the Open WebUI backend host. author: Thomas Friedel -version: 0.1 +contributors: Speedliner +version: 0.2 license: MIT requirements: httpx>=0.27 """ From 7e48b82b22078b3c09ae20a0baa6492bbb58469b Mon Sep 17 00:00:00 2001 From: Claudius Magicus Date: Fri, 7 Aug 2026 16:02:34 +0200 Subject: [PATCH 07/10] fix: repair IndentationError in claude_agent_pipe.py import block The rate-limit-surfacing feature (RateLimitEvent handling) added in ee6fc4c/4387685 left a stray leading space on the claude_agent_sdk import block, causing a module-level IndentationError that broke the whole pipe. Also cleans up two orphaned whitespace-only lines left between the RateLimitEvent and ResultMessage branches. --- claude_agent_pipe.py | 31 +++++++++++++++---------------- 1 file changed, 15 insertions(+), 16 deletions(-) diff --git a/claude_agent_pipe.py b/claude_agent_pipe.py index 7fc4019..426466c 100644 --- a/claude_agent_pipe.py +++ b/claude_agent_pipe.py @@ -44,20 +44,20 @@ _ARTIFACT_EXTENSIONS = _IMAGE_EXTENSIONS | _DOWNLOAD_EXTENSIONS # when large — this is only a "don't accidentally ship a DVD ISO" guard. _MAX_ARTIFACT_BYTES = 50 * 1024 * 1024 # 50 MiB - from claude_agent_sdk import ( - AssistantMessage, - ClaudeAgentOptions, - ClaudeSDKClient, - RateLimitEvent, - ResultMessage, - StreamEvent, - SystemMessage, - ToolResultBlock, - ToolUseBlock, - UserMessage, - create_sdk_mcp_server, - tool, - ) +from claude_agent_sdk import ( + AssistantMessage, + ClaudeAgentOptions, + ClaudeSDKClient, + RateLimitEvent, + ResultMessage, + StreamEvent, + SystemMessage, + ToolResultBlock, + ToolUseBlock, + UserMessage, + create_sdk_mcp_server, + tool, +) log = logging.getLogger(__name__) @@ -1724,8 +1724,7 @@ class Pipe: if info.status in ("allowed_warning", "rejected"): yield f"\n\n_{line}_\n" continue - - + if isinstance(message, ResultMessage): await emit_status("Done.", done=True) for chunk in _inline_new_artifacts( From 8f0d99100b69022547b90db3e1d3cb4102561cda Mon Sep 17 00:00:00 2001 From: Claudius Magicus Date: Fri, 7 Aug 2026 16:09:12 +0200 Subject: [PATCH 08/10] chore: strip orphaned whitespace-only line in rate-limit event handler Cosmetic cleanup of a leftover whitespace line after the rate_limit_event branch in _handle_event (from 6ac8d3b), no functional change. --- claude_agent_pipe_sandboxed.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/claude_agent_pipe_sandboxed.py b/claude_agent_pipe_sandboxed.py index 3c77df7..eca1474 100644 --- a/claude_agent_pipe_sandboxed.py +++ b/claude_agent_pipe_sandboxed.py @@ -1021,7 +1021,7 @@ async def _handle_event( if status in ("allowed_warning", "rejected"): yield f"\n\n_{line}_\n" return - + if etype == "result": subtype = event.get("subtype") if subtype and subtype != "success": From 3c2217115be54962cfe4519d67cb771587887acf Mon Sep 17 00:00:00 2001 From: Assistant Patch Date: Mon, 10 Aug 2026 13:21:36 +0000 Subject: [PATCH 09/10] Fix: persist Claude Code session id across process restarts and workers - Add session_store.py: SQLite-backed, WAL-mode session store keyed by chat_id, robust to multi-worker access and process restarts. - Add session_marker.py: fallback mechanism that embeds an invisible markdown reference marker in assistant responses so the session id survives even without SQLite access, by parsing it back out of body messages history. - Update claude_agent_pipe.py: resolution order is in-memory cache -> SQLite store -> marker fallback. Session id is persisted to both the in-memory dict and SQLite on every init SystemMessage, and the marker is appended to the final visible response text. - Warn when chat_id is missing/None instead of silently starting a fresh session (related to open-webui/open-webui#20563). --- claude_agent_pipe.py | 65 ++++++++++++++++++++- session_marker.py | 96 +++++++++++++++++++++++++++++++ session_store.py | 134 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 293 insertions(+), 2 deletions(-) create mode 100644 session_marker.py create mode 100644 session_store.py diff --git a/claude_agent_pipe.py b/claude_agent_pipe.py index 426466c..34ab0ec 100644 --- a/claude_agent_pipe.py +++ b/claude_agent_pipe.py @@ -21,6 +21,9 @@ from typing import Any, AsyncGenerator, Callable, Dict, List, Optional, Set from pydantic import BaseModel, Field +from session_store import get_store +from session_marker import extract_session_id_from_messages, make_marker + _IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp"} _DOWNLOAD_EXTENSIONS = { ".pdf", @@ -61,8 +64,15 @@ from claude_agent_sdk import ( log = logging.getLogger(__name__) -# OpenWebUI calls pipe() fresh for each chat turn. We keep a chat_id -> session_id -# map in-process so follow-up turns resume the same Claude Code session. +# OpenWebUI calls pipe() fresh for each chat turn. This in-process dict is a +# fast path so a hot worker doesn't have to hit SQLite on every turn. It is +# NOT the source of truth anymore: it starts empty on every process restart +# and is per-worker (stale/absent on any other worker), which is exactly why +# Claude Code used to "forget" chat history across restarts / multi-worker +# deployments. The source of truth is session_store.SessionStore (SQLite, +# survives restarts, shared across workers via the shared filesystem), with +# session_marker as a secondary fallback embedded in the chat text itself. +# See session_store.py / session_marker.py for the full rationale. _chat_sessions: Dict[str, str] = {} @@ -1495,13 +1505,49 @@ class Pipe: prompt = _strip_mode_prefix(prompt) chat_id = __chat_id__ or "default" + if not __chat_id__: + log.warning( + "pipe() called without __chat_id__ — falling back to a shared " + "'default' workdir/session. Session resume across turns will " + "not work correctly for this call (see OpenWebUI issue about " + "metadata.chat_id being unset for certain internal calls)." + ) workdir = Path(self.valves.WORKDIR_ROOT) / chat_id workdir.mkdir(parents=True, exist_ok=True) allowed_tools = [ t.strip() for t in self.valves.ALLOWED_TOOLS.split(",") if t.strip() ] + + # Resolve the session to resume, in order of preference: + # 1. In-process cache (_chat_sessions) — fastest, but empty after a + # restart or on any worker that didn't handle the previous turn. + # 2. SQLite-backed SessionStore — survives restarts and is shared + # across worker processes via the shared filesystem under + # WORKDIR_ROOT. This is the source of truth. + # 3. Marker embedded in the previous assistant message, extracted + # from OpenWebUI's own chat history in `body`. Fallback only, + # for the case where the SQLite file itself became unavailable + # (e.g. WORKDIR_ROOT lives on ephemeral storage that was wiped + # independently of OpenWebUI's own chat database). + session_store = get_store(self.valves.WORKDIR_ROOT) resume_id = _chat_sessions.get(chat_id) + resume_source = "memory" if resume_id else None + if not resume_id: + resume_id = session_store.get(chat_id) + if resume_id: + resume_source = "sqlite" + if not resume_id: + resume_id = extract_session_id_from_messages(body.get("messages") or []) + if resume_id: + resume_source = "marker" + if resume_id: + log.debug( + "Resuming Claude Code session %s for chat_id=%s (source=%s)", + resume_id, + chat_id, + resume_source, + ) # Knowledge base attached via Workspace Model → expose as an MCP tool # Claude can call agentically. OpenWebUI's middleware already added one @@ -1621,6 +1667,11 @@ class Pipe: session_id = message.data.get("session_id") if session_id: _chat_sessions[chat_id] = session_id + # Persist beyond this process's lifetime/worker. + # Cheap (single upsert) and done on every init + # message, i.e. once per turn — not hot-path + # sensitive. + session_store.set(chat_id, session_id) continue if isinstance(message, StreamEvent): @@ -1739,6 +1790,16 @@ class Pipe: yield f"\n\n**API-Fehler:** `{message.api_error_status}`\n" if message.total_cost_usd is not None: yield f"\n\n_Cost: ${message.total_cost_usd:.4f} · {message.duration_ms}ms_\n" + # Fallback-path marker (see session_marker.py): embed + # the session_id invisibly in the reply itself so it + # can be recovered from OpenWebUI's own chat history + # even if both the in-memory cache and the SQLite + # store are unavailable on a later turn. This does + # NOT write into any OpenWebUI-owned data — it's part + # of the ordinary streamed assistant text. + current_session_id = _chat_sessions.get(chat_id) + if current_session_id: + yield make_marker(current_session_id) return except Exception as exc: diff --git a/session_marker.py b/session_marker.py new file mode 100644 index 0000000..714c435 --- /dev/null +++ b/session_marker.py @@ -0,0 +1,96 @@ +""" +Invisible in-text marker carrying the Claude Code session_id as a fallback +to the SQLite-backed SessionStore (see session_store.py). + +Rationale +--------- +The SQLite store (Variant A) is the primary source of truth: it survives +process restarts and worker changes without depending on what OpenWebUI +sends back in the request body. But it has one failure mode worth guarding +against: if WORKDIR_ROOT points at ephemeral/container-local storage that +gets wiped independently of OpenWebUI's own chat database (e.g. different +persistent-volume lifecycle in a container redeploy), the SQLite file can +disappear while OpenWebUI's chat history — which is persisted separately and +NOT touched by this pipe — still exists and still contains the last +assistant reply. + +To cover that case, every assistant reply gets an invisible marker appended: + + [claude-session:]: # + +This is a Markdown *link reference definition* (CommonMark spec, "reference +link" section): a line of the form `[label]: destination` that is never +rendered as visible output by CommonMark-compliant renderers — it just +registers a reference. Using `#` as the destination keeps it inert (no +actual link target semantics matter here; we only care about it being +absent from rendered output). This is the same technique already validated +in production by rbb-dev/Open-WebUI-OpenRouter-pipe for larger artifact +references. + +Important: this marker rides inside the ordinary assistant message text +that OpenWebUI already persists as part of the normal chat flow. Nothing is +written into OpenWebUI's own chat/meta database columns from here — there is +no additional write path into OpenWebUI's data model at all, so there is no +extra conflict/race-condition surface beyond what streaming a normal +response already has. + +Limitations (see also the write-up in chat): this only works if OpenWebUI +resends the previous assistant message inside `body["messages"]` on the next +turn (the normal case for OpenAI-Chat-Completions-shaped pipes). It is a +fallback, not a replacement, for the SQLite store. +""" + +from __future__ import annotations + +import re +from typing import Any, Dict, List, Optional + +_MARKER_RE = re.compile(r"\[claude-session:([A-Za-z0-9_-]{1,128})\]:\s*#") + + +def make_marker(session_id: str) -> str: + """Return the invisible marker text to append to a streamed reply.""" + return f"\n\n[claude-session:{session_id}]: #\n" + + +def extract_session_id_from_messages(messages: List[Dict[str, Any]]) -> Optional[str]: + """Scan chat history (most recent first) for the last embedded marker. + + Looks only at assistant messages, newest to oldest, and returns the + session_id from the first marker found. Returns None if no marker is + present (e.g. first turn in a chat, or history was trimmed/edited). + """ + for message in reversed(messages or []): + if message.get("role") != "assistant": + continue + content = message.get("content") + text = _flatten_content(content) + if not text: + continue + match = _MARKER_RE.search(text) + if match: + return match.group(1) + return None + + +def strip_marker(text: str) -> str: + """Remove marker lines from text before showing/logging it elsewhere. + + Not needed for normal rendering (CommonMark already hides it), but + useful if the raw text is re-used somewhere that doesn't apply Markdown + rendering (e.g. plain-text export, logs). + """ + return _MARKER_RE.sub("", text) + + +def _flatten_content(content: Any) -> str: + if isinstance(content, str): + return content + if isinstance(content, list): + parts = [ + part.get("text", "") + for part in content + if isinstance(part, dict) and part.get("type") == "text" + ] + return "\n".join(parts) + return "" diff --git a/session_store.py b/session_store.py new file mode 100644 index 0000000..32de925 --- /dev/null +++ b/session_store.py @@ -0,0 +1,134 @@ +""" +Persistent chat_id -> Claude Code session_id mapping. + +Why this exists +---------------- +`claude_agent_pipe.py` used to keep this mapping in a plain in-process dict +(`_chat_sessions`). That breaks in three common situations: + + 1. The backend process restarts (redeploy, crash, admin reloads the + function) -> the dict is empty again. + 2. The backend runs with more than one worker process / replica -> each + worker has its own dict, so whichever worker handles the next turn may + simply not know about the session the previous turn created. + 3. Long-lived deployments accumulate sessions for chats that are no longer + active with no way to expire them. + +This module replaces the in-memory dict with a small SQLite database that +lives next to the per-chat workdir (`WORKDIR_ROOT/.session_store.sqlite3`). +It is intentionally NOT integrated with OpenWebUI's own database: OpenWebUI's +internal SQLAlchemy engine/schema is not a stable, documented plugin API, and +writing into OpenWebUI's own `chat`/`message` tables risks racing with +OpenWebUI's own read-modify-write cycle on the same rows. A separate SQLite +file has no such overlap and needs no coordination with OpenWebUI at all. + +SQLite is a reasonable choice here (rather than e.g. requiring Redis) because +the write volume is tiny (one row write per chat turn) and SQLite's built-in +locking is sufficient for the "one active turn per chat_id at a time" access +pattern this pipe has. If you run many worker processes hammering the *same* +chat_id concurrently, SQLite's default locking will simply serialize those +writes rather than corrupt anything. + +Usage +----- + store = SessionStore(root_dir) + resume_id = store.get(chat_id) + ... + store.set(chat_id, session_id) +""" + +from __future__ import annotations + +import logging +import sqlite3 +import threading +import time +from pathlib import Path +from typing import Optional + +log = logging.getLogger(__name__) + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS claude_code_sessions ( + chat_id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + updated_at REAL NOT NULL +); +""" + + +class SessionStore: + """Thread-safe, file-backed chat_id -> session_id store. + + One SQLite connection per instance, guarded by a lock. Pipes are invoked + concurrently for different chats within the same process, so the lock is + held only for the duration of a single get/set (a few milliseconds), + not for the whole agent turn. + """ + + def __init__(self, root_dir: str | Path, filename: str = ".session_store.sqlite3") -> None: + self._path = Path(root_dir) / filename + self._path.parent.mkdir(parents=True, exist_ok=True) + self._lock = threading.Lock() + # check_same_thread=False: pipe() runs inside asyncio, potentially + # scheduled on different threads across calls (e.g. via run_in_executor + # elsewhere in the codebase). Access is still serialized by self._lock. + self._conn = sqlite3.connect(str(self._path), check_same_thread=False, timeout=10) + self._conn.execute("PRAGMA journal_mode=WAL;") + self._conn.execute("PRAGMA busy_timeout=5000;") + with self._lock: + self._conn.execute(_SCHEMA) + self._conn.commit() + + def get(self, chat_id: str) -> Optional[str]: + try: + with self._lock: + cur = self._conn.execute( + "SELECT session_id FROM claude_code_sessions WHERE chat_id = ?", + (chat_id,), + ) + row = cur.fetchone() + return row[0] if row else None + except sqlite3.Error: + log.exception("SessionStore.get failed for chat_id=%s", chat_id) + return None + + def set(self, chat_id: str, session_id: str) -> None: + try: + with self._lock: + self._conn.execute( + """ + INSERT INTO claude_code_sessions (chat_id, session_id, updated_at) + VALUES (?, ?, ?) + ON CONFLICT(chat_id) DO UPDATE SET + session_id = excluded.session_id, + updated_at = excluded.updated_at + """, + (chat_id, session_id, time.time()), + ) + self._conn.commit() + except sqlite3.Error: + log.exception("SessionStore.set failed for chat_id=%s", chat_id) + + def close(self) -> None: + with self._lock: + self._conn.close() + + +_stores: dict[str, SessionStore] = {} +_stores_lock = threading.Lock() + + +def get_store(root_dir: str | Path) -> SessionStore: + """Return a process-wide singleton SessionStore per WORKDIR_ROOT. + + Avoids opening a new SQLite connection on every single pipe() call while + still keying correctly if the valve ever points at a different root. + """ + key = str(root_dir) + with _stores_lock: + store = _stores.get(key) + if store is None: + store = SessionStore(root_dir) + _stores[key] = store + return store From 86933746aa64cdc2c3a4dcbb10233c7524c2637d Mon Sep 17 00:00:00 2001 From: Assistant Patch Date: Mon, 10 Aug 2026 20:45:48 +0000 Subject: [PATCH 10/10] Fix: inline session_store/session_marker into claude_agent_pipe.py OpenWebUI loads Functions as a single in-memory module via exec() (open_webui/utils/plugin.py: load_function_module_by_id). There is no mechanism to ship sibling .py files alongside a Function, so 'from session_store import get_store' / 'from session_marker import ...' fail with ModuleNotFoundError as soon as the function is pasted/imported into OpenWebUI (see traceback: exec(content, module.__dict__) -> line 24 -> ModuleNotFoundError: No module named 'session_store'). Fix: inline both modules' full content directly into claude_agent_pipe.py so it is fully self-contained, and remove the now redundant standalone session_store.py / session_marker.py files (their logic lives inline now; keeping both would only invite drift). Verified by: - python3 -m py_compile claude_agent_pipe.py - exec()'ing the file the same way OpenWebUI's plugin loader does (with claude_agent_sdk stubbed out), confirming no ModuleNotFoundError/NameError and that SessionStore/get_store/make_marker/extract_session_id_from_messages and the Pipe class all load correctly. --- claude_agent_pipe.py | 161 ++++++++++++++++++++++++++++++++++++++++++- session_marker.py | 96 -------------------------- session_store.py | 134 ----------------------------------- 3 files changed, 159 insertions(+), 232 deletions(-) delete mode 100644 session_marker.py delete mode 100644 session_store.py diff --git a/claude_agent_pipe.py b/claude_agent_pipe.py index 34ab0ec..0fd949e 100644 --- a/claude_agent_pipe.py +++ b/claude_agent_pipe.py @@ -21,8 +21,165 @@ from typing import Any, AsyncGenerator, Callable, Dict, List, Optional, Set from pydantic import BaseModel, Field -from session_store import get_store -from session_marker import extract_session_id_from_messages, make_marker +# --------------------------------------------------------------------------- +# Inlined from session_store.py / session_marker.py. +# +# OpenWebUI Functions are loaded as a single in-memory module via exec() (see +# open_webui/utils/plugin.py: load_function_module_by_id) — there is no +# mechanism to ship additional .py files alongside a Function, so importing +# sibling modules (`from session_store import ...`) fails with +# ModuleNotFoundError at install time. Both modules are therefore inlined +# here verbatim. Keep the original files in the repo in sync if you edit +# this logic — they are the canonical, independently testable source; this +# inlined copy is what actually ships to OpenWebUI. +# --------------------------------------------------------------------------- + +import sqlite3 +import threading + +_SESSION_STORE_SCHEMA = """ +CREATE TABLE IF NOT EXISTS claude_code_sessions ( + chat_id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + updated_at REAL NOT NULL +); +""" + + +class SessionStore: + """Thread-safe, file-backed chat_id -> session_id store. + + Persists the chat_id -> Claude Code session_id mapping in a small SQLite + database next to the per-chat workdir (WORKDIR_ROOT/.session_store.sqlite3). + This replaces a plain in-process dict, which loses all mappings on every + backend restart and is inconsistent across multiple worker processes. + + Deliberately NOT integrated with OpenWebUI's own database: OpenWebUI's + internal SQLAlchemy engine/schema is not a stable, documented plugin API, + and writing into OpenWebUI's own chat/message tables risks racing with + OpenWebUI's own read-modify-write cycle on those rows. A separate SQLite + file has no such overlap and needs no coordination with OpenWebUI. + """ + + def __init__(self, root_dir, filename: str = ".session_store.sqlite3") -> None: + self._path = Path(root_dir) / filename + self._path.parent.mkdir(parents=True, exist_ok=True) + self._lock = threading.Lock() + self._conn = sqlite3.connect(str(self._path), check_same_thread=False, timeout=10) + self._conn.execute("PRAGMA journal_mode=WAL;") + self._conn.execute("PRAGMA busy_timeout=5000;") + with self._lock: + self._conn.execute(_SESSION_STORE_SCHEMA) + self._conn.commit() + + def get(self, chat_id: str) -> Optional[str]: + try: + with self._lock: + cur = self._conn.execute( + "SELECT session_id FROM claude_code_sessions WHERE chat_id = ?", + (chat_id,), + ) + row = cur.fetchone() + return row[0] if row else None + except sqlite3.Error: + log.exception("SessionStore.get failed for chat_id=%s", chat_id) + return None + + def set(self, chat_id: str, session_id: str) -> None: + try: + with self._lock: + self._conn.execute( + """ + INSERT INTO claude_code_sessions (chat_id, session_id, updated_at) + VALUES (?, ?, ?) + ON CONFLICT(chat_id) DO UPDATE SET + session_id = excluded.session_id, + updated_at = excluded.updated_at + """, + (chat_id, session_id, time.time()), + ) + self._conn.commit() + except sqlite3.Error: + log.exception("SessionStore.set failed for chat_id=%s", chat_id) + + def close(self) -> None: + with self._lock: + self._conn.close() + + +_session_stores: Dict[str, "SessionStore"] = {} +_session_stores_lock = threading.Lock() + + +def get_store(root_dir) -> "SessionStore": + """Return a process-wide singleton SessionStore per WORKDIR_ROOT.""" + key = str(root_dir) + with _session_stores_lock: + store = _session_stores.get(key) + if store is None: + store = SessionStore(root_dir) + _session_stores[key] = store + return store + + +_SESSION_MARKER_RE = re.compile(r"\[claude-session:([A-Za-z0-9_-]{1,128})\]:\s*#") + + +def make_marker(session_id: str) -> str: + """Invisible in-text marker carrying the session_id, appended to replies. + + Fallback for when the SQLite store is unreachable/wiped independently of + OpenWebUI's own chat history (e.g. different volume lifecycle across a + container redeploy). This is a Markdown link reference definition + (`[label]: #`), which CommonMark-compliant renderers register but never + render as visible output. Rides inside the ordinary assistant message + text that OpenWebUI already persists — no extra write path into + OpenWebUI's own data model, hence no extra race-condition surface. + """ + return f"\n\n[claude-session:{session_id}]: #\n" + + +def extract_session_id_from_messages(messages: List[Dict[str, Any]]) -> Optional[str]: + """Scan chat history (most recent first) for the last embedded marker. + + Only works if OpenWebUI resends prior assistant messages in + body["messages"] (the normal case for Chat-Completions-shaped pipes). + Fallback only, not a replacement for the SQLite store. + """ + for message in reversed(messages or []): + if message.get("role") != "assistant": + continue + content = message.get("content") + text = _flatten_marker_content(content) + if not text: + continue + match = _SESSION_MARKER_RE.search(text) + if match: + return match.group(1) + return None + + +def strip_marker(text: str) -> str: + """Remove marker lines from text before showing/logging it elsewhere.""" + return _SESSION_MARKER_RE.sub("", text) + + +def _flatten_marker_content(content: Any) -> str: + if isinstance(content, str): + return content + if isinstance(content, list): + parts = [ + part.get("text", "") + for part in content + if isinstance(part, dict) and part.get("type") == "text" + ] + return "\n".join(parts) + return "" + + +# --------------------------------------------------------------------------- +# End of inlined session_store.py / session_marker.py +# --------------------------------------------------------------------------- _IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp"} _DOWNLOAD_EXTENSIONS = { diff --git a/session_marker.py b/session_marker.py deleted file mode 100644 index 714c435..0000000 --- a/session_marker.py +++ /dev/null @@ -1,96 +0,0 @@ -""" -Invisible in-text marker carrying the Claude Code session_id as a fallback -to the SQLite-backed SessionStore (see session_store.py). - -Rationale ---------- -The SQLite store (Variant A) is the primary source of truth: it survives -process restarts and worker changes without depending on what OpenWebUI -sends back in the request body. But it has one failure mode worth guarding -against: if WORKDIR_ROOT points at ephemeral/container-local storage that -gets wiped independently of OpenWebUI's own chat database (e.g. different -persistent-volume lifecycle in a container redeploy), the SQLite file can -disappear while OpenWebUI's chat history — which is persisted separately and -NOT touched by this pipe — still exists and still contains the last -assistant reply. - -To cover that case, every assistant reply gets an invisible marker appended: - - [claude-session:]: # - -This is a Markdown *link reference definition* (CommonMark spec, "reference -link" section): a line of the form `[label]: destination` that is never -rendered as visible output by CommonMark-compliant renderers — it just -registers a reference. Using `#` as the destination keeps it inert (no -actual link target semantics matter here; we only care about it being -absent from rendered output). This is the same technique already validated -in production by rbb-dev/Open-WebUI-OpenRouter-pipe for larger artifact -references. - -Important: this marker rides inside the ordinary assistant message text -that OpenWebUI already persists as part of the normal chat flow. Nothing is -written into OpenWebUI's own chat/meta database columns from here — there is -no additional write path into OpenWebUI's data model at all, so there is no -extra conflict/race-condition surface beyond what streaming a normal -response already has. - -Limitations (see also the write-up in chat): this only works if OpenWebUI -resends the previous assistant message inside `body["messages"]` on the next -turn (the normal case for OpenAI-Chat-Completions-shaped pipes). It is a -fallback, not a replacement, for the SQLite store. -""" - -from __future__ import annotations - -import re -from typing import Any, Dict, List, Optional - -_MARKER_RE = re.compile(r"\[claude-session:([A-Za-z0-9_-]{1,128})\]:\s*#") - - -def make_marker(session_id: str) -> str: - """Return the invisible marker text to append to a streamed reply.""" - return f"\n\n[claude-session:{session_id}]: #\n" - - -def extract_session_id_from_messages(messages: List[Dict[str, Any]]) -> Optional[str]: - """Scan chat history (most recent first) for the last embedded marker. - - Looks only at assistant messages, newest to oldest, and returns the - session_id from the first marker found. Returns None if no marker is - present (e.g. first turn in a chat, or history was trimmed/edited). - """ - for message in reversed(messages or []): - if message.get("role") != "assistant": - continue - content = message.get("content") - text = _flatten_content(content) - if not text: - continue - match = _MARKER_RE.search(text) - if match: - return match.group(1) - return None - - -def strip_marker(text: str) -> str: - """Remove marker lines from text before showing/logging it elsewhere. - - Not needed for normal rendering (CommonMark already hides it), but - useful if the raw text is re-used somewhere that doesn't apply Markdown - rendering (e.g. plain-text export, logs). - """ - return _MARKER_RE.sub("", text) - - -def _flatten_content(content: Any) -> str: - if isinstance(content, str): - return content - if isinstance(content, list): - parts = [ - part.get("text", "") - for part in content - if isinstance(part, dict) and part.get("type") == "text" - ] - return "\n".join(parts) - return "" diff --git a/session_store.py b/session_store.py deleted file mode 100644 index 32de925..0000000 --- a/session_store.py +++ /dev/null @@ -1,134 +0,0 @@ -""" -Persistent chat_id -> Claude Code session_id mapping. - -Why this exists ----------------- -`claude_agent_pipe.py` used to keep this mapping in a plain in-process dict -(`_chat_sessions`). That breaks in three common situations: - - 1. The backend process restarts (redeploy, crash, admin reloads the - function) -> the dict is empty again. - 2. The backend runs with more than one worker process / replica -> each - worker has its own dict, so whichever worker handles the next turn may - simply not know about the session the previous turn created. - 3. Long-lived deployments accumulate sessions for chats that are no longer - active with no way to expire them. - -This module replaces the in-memory dict with a small SQLite database that -lives next to the per-chat workdir (`WORKDIR_ROOT/.session_store.sqlite3`). -It is intentionally NOT integrated with OpenWebUI's own database: OpenWebUI's -internal SQLAlchemy engine/schema is not a stable, documented plugin API, and -writing into OpenWebUI's own `chat`/`message` tables risks racing with -OpenWebUI's own read-modify-write cycle on the same rows. A separate SQLite -file has no such overlap and needs no coordination with OpenWebUI at all. - -SQLite is a reasonable choice here (rather than e.g. requiring Redis) because -the write volume is tiny (one row write per chat turn) and SQLite's built-in -locking is sufficient for the "one active turn per chat_id at a time" access -pattern this pipe has. If you run many worker processes hammering the *same* -chat_id concurrently, SQLite's default locking will simply serialize those -writes rather than corrupt anything. - -Usage ------ - store = SessionStore(root_dir) - resume_id = store.get(chat_id) - ... - store.set(chat_id, session_id) -""" - -from __future__ import annotations - -import logging -import sqlite3 -import threading -import time -from pathlib import Path -from typing import Optional - -log = logging.getLogger(__name__) - -_SCHEMA = """ -CREATE TABLE IF NOT EXISTS claude_code_sessions ( - chat_id TEXT PRIMARY KEY, - session_id TEXT NOT NULL, - updated_at REAL NOT NULL -); -""" - - -class SessionStore: - """Thread-safe, file-backed chat_id -> session_id store. - - One SQLite connection per instance, guarded by a lock. Pipes are invoked - concurrently for different chats within the same process, so the lock is - held only for the duration of a single get/set (a few milliseconds), - not for the whole agent turn. - """ - - def __init__(self, root_dir: str | Path, filename: str = ".session_store.sqlite3") -> None: - self._path = Path(root_dir) / filename - self._path.parent.mkdir(parents=True, exist_ok=True) - self._lock = threading.Lock() - # check_same_thread=False: pipe() runs inside asyncio, potentially - # scheduled on different threads across calls (e.g. via run_in_executor - # elsewhere in the codebase). Access is still serialized by self._lock. - self._conn = sqlite3.connect(str(self._path), check_same_thread=False, timeout=10) - self._conn.execute("PRAGMA journal_mode=WAL;") - self._conn.execute("PRAGMA busy_timeout=5000;") - with self._lock: - self._conn.execute(_SCHEMA) - self._conn.commit() - - def get(self, chat_id: str) -> Optional[str]: - try: - with self._lock: - cur = self._conn.execute( - "SELECT session_id FROM claude_code_sessions WHERE chat_id = ?", - (chat_id,), - ) - row = cur.fetchone() - return row[0] if row else None - except sqlite3.Error: - log.exception("SessionStore.get failed for chat_id=%s", chat_id) - return None - - def set(self, chat_id: str, session_id: str) -> None: - try: - with self._lock: - self._conn.execute( - """ - INSERT INTO claude_code_sessions (chat_id, session_id, updated_at) - VALUES (?, ?, ?) - ON CONFLICT(chat_id) DO UPDATE SET - session_id = excluded.session_id, - updated_at = excluded.updated_at - """, - (chat_id, session_id, time.time()), - ) - self._conn.commit() - except sqlite3.Error: - log.exception("SessionStore.set failed for chat_id=%s", chat_id) - - def close(self) -> None: - with self._lock: - self._conn.close() - - -_stores: dict[str, SessionStore] = {} -_stores_lock = threading.Lock() - - -def get_store(root_dir: str | Path) -> SessionStore: - """Return a process-wide singleton SessionStore per WORKDIR_ROOT. - - Avoids opening a new SQLite connection on every single pipe() call while - still keying correctly if the valve ever points at a different root. - """ - key = str(root_dir) - with _stores_lock: - store = _stores.get(key) - if store is None: - store = SessionStore(root_dir) - _stores[key] = store - return store