fix: update stale docstring referencing removed _MCPPathRewriteMiddleware

Co-Authored-By: Claude Opus 4.6 <[email protected]>

hindsight-api/hindsight_api/api/__init__.py

@@ -6,7 +6,6 @@ Provides both HTTP REST API and MCP (Model Context Protocol) server.
 
 import logging
 from contextlib import asynccontextmanager
-from typing import Optional
 
 from fastapi import FastAPI
 
@@ -46,14 +45,14 @@ def create_app(
         # Both HTTP and MCP
         app = create_app(memory, mcp_api_enabled=True)
     """
-    mcp_app = None
+    mcp_servers = None
 
-    # Create MCP app first if enabled (we need its lifespan for chaining)
+    # Create MCP servers first if enabled (we need their lifespans for chaining)
     if mcp_api_enabled:
         try:
-            from .mcp import create_mcp_app
+            from .mcp import MCPMiddleware, create_mcp_servers
 
-            mcp_app = create_mcp_app(memory=memory)
+            mcp_servers = create_mcp_servers(memory=memory)
         except ImportError as e:
             logger.error(f"MCP server requested but dependencies not available: {e}")
             logger.error("Install with: pip install hindsight-api[mcp]")
@@ -70,11 +69,9 @@ def create_app(
         app = FastAPI(title="Hindsight API", version="0.0.7")
         logger.info("HTTP REST API disabled")
 
-    # Mount MCP server and chain its lifespan if enabled
-    if mcp_app is not None:
-        # Get both MCP apps' underlying Starlette apps for lifespan access
-        multi_bank_starlette_app = mcp_app.multi_bank_app
-        single_bank_starlette_app = mcp_app.single_bank_app
+    # Add MCP middleware and chain its lifespan if enabled
+    if mcp_servers is not None:
+        multi_bank_server, single_bank_server, multi_bank_starlette_app, single_bank_starlette_app = mcp_servers
 
         # Store the original lifespan
         original_lifespan = app.router.lifespan_context
@@ -94,8 +91,19 @@ def create_app(
         # Replace the app's lifespan with the chained version
         app.router.lifespan_context = chained_lifespan
 
-        # Mount the MCP middleware
-        app.mount(mcp_mount_path, mcp_app)
+        # Add MCP as a wrapping middleware — intercepts /mcp* requests directly,
+        # passes everything else through to the FastAPI app. No Starlette Mount
+        # means no 307 redirect for /mcp (no trailing slash).
+        app.add_middleware(
+            MCPMiddleware,
+            memory=memory,
+            prefix=mcp_mount_path,
+            multi_bank_app=multi_bank_starlette_app,
+            single_bank_app=single_bank_starlette_app,
+            multi_bank_server=multi_bank_server,
+            single_bank_server=single_bank_server,
+        )
+
         logger.info(f"MCP server enabled at {mcp_mount_path}/")
 
     return app

hindsight-api/hindsight_api/api/mcp.py

@@ -118,7 +118,10 @@ def create_mcp_server(memory: MemoryEngine, multi_bank: bool = True) -> FastMCP:
 
 
 class MCPMiddleware:
-    """ASGI middleware that handles authentication and routes to appropriate MCP server.
+    """ASGI middleware that intercepts MCP requests and routes to appropriate MCP server.
+
+    This middleware wraps the main FastAPI app and intercepts requests matching the
+    configured prefix (default: /mcp). Non-MCP requests pass through to the inner app.
 
     Authentication:
         1. If HINDSIGHT_API_MCP_AUTH_TOKEN is set (legacy), validates against that token
@@ -149,27 +152,33 @@ class MCPMiddleware:
             --header "X-Bank-Id: my-bank" --header "Authorization: Bearer <token>"
     """
 
-    def __init__(self, app, memory: MemoryEngine):
+    def __init__(
+        self,
+        app,
+        memory: MemoryEngine,
+        prefix: str = "/mcp",
+        multi_bank_app=None,
+        single_bank_app=None,
+        multi_bank_server=None,
+        single_bank_server=None,
+    ):
         self.app = app
+        self.prefix = prefix
         self.memory = memory
         self.tenant_extension = memory._tenant_extension
 
-        # Create two server instances:
-        # 1. Multi-bank server (for /mcp/ root endpoint)
-        self.multi_bank_server = create_mcp_server(memory, multi_bank=True)
-        self.multi_bank_app = self.multi_bank_server.http_app(path="/")
-
-        # 2. Single-bank server (for /mcp/{bank_id}/ endpoints)
-        self.single_bank_server = create_mcp_server(memory, multi_bank=False)
-        self.single_bank_app = self.single_bank_server.http_app(path="/")
-
-        # Backward compatibility: expose multi_bank_app as mcp_app
-        self.mcp_app = self.multi_bank_app
-
-        # Expose the lifespan for the parent app to chain (use multi-bank as default)
-        self.lifespan = (
-            self.multi_bank_app.lifespan_handler if hasattr(self.multi_bank_app, "lifespan_handler") else None
-        )
+        if multi_bank_app and single_bank_app:
+            # Pre-created servers (used when called via add_middleware from create_app)
+            self.multi_bank_app = multi_bank_app
+            self.single_bank_app = single_bank_app
+            self.multi_bank_server = multi_bank_server
+            self.single_bank_server = single_bank_server
+        else:
+            # Create servers internally (for direct construction / tests)
+            self.multi_bank_server = create_mcp_server(memory, multi_bank=True)
+            self.multi_bank_app = self.multi_bank_server.http_app(path="/")
+            self.single_bank_server = create_mcp_server(memory, multi_bank=False)
+            self.single_bank_app = self.single_bank_server.http_app(path="/")
 
     def _get_header(self, scope: dict, name: str) -> str | None:
         """Extract a header value from ASGI scope."""
@@ -181,9 +190,20 @@ class MCPMiddleware:
 
     async def __call__(self, scope, receive, send):
         if scope["type"] != "http":
-            await self.multi_bank_app(scope, receive, send)
+            await self.app(scope, receive, send)
             return
 
+        path = scope.get("path", "")
+
+        # Check if this is an MCP request (matches prefix)
+        if not (path == self.prefix or path.startswith(self.prefix + "/")):
+            # Not an MCP request — pass through to the inner app
+            await self.app(scope, receive, send)
+            return
+
+        # Strip prefix from path
+        path = path[len(self.prefix) :] or "/"
+
         # Extract auth token from header (for tenant auth propagation)
         auth_header = self._get_header(scope, "Authorization")
         auth_token: str | None = None
@@ -222,36 +242,15 @@ class MCPMiddleware:
             _current_schema.set(tenant_context.schema_name) if tenant_context and tenant_context.schema_name else None
         )
 
-        path = scope.get("path", "")
-
-        # Strip any mount prefix (e.g., /mcp) that FastAPI might not have stripped
-        root_path = scope.get("root_path", "")
-        if root_path and path.startswith(root_path):
-            path = path[len(root_path) :] or "/"
-
-        # Also handle case where mount path wasn't stripped (e.g., /mcp/...)
-        if path.startswith("/mcp/"):
-            path = path[4:]  # Remove /mcp prefix
-        elif path == "/mcp":
-            path = "/"
-
-        # Ensure path has leading slash (needed after stripping mount path)
-        if path and not path.startswith("/"):
-            path = "/" + path
-
         # Try to get bank_id from header first (for Claude Code compatibility)
         bank_id = self._get_header(scope, "X-Bank-Id")
         bank_id_from_path = False
 
-        # MCP endpoint paths that should not be treated as bank_ids
-        MCP_ENDPOINTS = {"sse", "messages"}
-
         # If no header, try to extract from path: /{bank_id}/...
         new_path = path
         if not bank_id and path.startswith("/") and len(path) > 1:
             parts = path[1:].split("/", 1)
-            # Don't treat MCP endpoints as bank_ids
-            if parts[0] and parts[0] not in MCP_ENDPOINTS:
+            if parts[0]:
                 # First segment looks like a bank_id
                 bank_id = parts[0]
                 bank_id_from_path = True
@@ -280,9 +279,19 @@ class MCPMiddleware:
             # Clear root_path since we're passing directly to the app
             new_scope["root_path"] = ""
 
-            # Wrap send to rewrite the SSE endpoint URL to include bank_id if using path-based routing
+            # Wrap send to rewrite the SSE endpoint URL to include bank_id if using path-based routing.
+            # Only rewrite SSE (text/event-stream) responses to avoid corrupting tool results
+            # that might contain the literal string "data: /messages".
+            is_sse_response = False
+
             async def send_wrapper(message):
-                if message["type"] == "http.response.body" and bank_id_from_path:
+                nonlocal is_sse_response
+                if message["type"] == "http.response.start":
+                    for header_name, header_value in message.get("headers", []):
+                        if header_name == b"content-type" and b"text/event-stream" in header_value:
+                            is_sse_response = True
+                            break
+                if message["type"] == "http.response.body" and bank_id_from_path and is_sse_response:
                     body = message.get("body", b"")
                     if body and b"/messages" in body:
                         # Rewrite /messages to /{bank_id}/messages in SSE endpoint event
@@ -320,30 +329,19 @@ class MCPMiddleware:
         )
 
 
-def create_mcp_app(memory: MemoryEngine):
-    """
-    Create an ASGI app that handles MCP requests with dynamic tool exposure.
+def create_mcp_servers(memory: MemoryEngine):
+    """Create multi-bank and single-bank MCP servers and their Starlette apps.
 
-    Authentication:
-        Uses the TenantExtension from the MemoryEngine (same auth as REST API).
-
-    Two modes based on URL structure:
-
-    1. Single-bank mode (recommended for agent isolation):
-       - URL: /mcp/{bank_id}/
-       - Tools: retain, recall, reflect (no bank_id parameter)
-       - Example: claude mcp add --transport http my-agent http://localhost:8888/mcp/my-agent-bank/
-
-    2. Multi-bank mode (for cross-bank operations):
-       - URL: /mcp/
-       - Tools: retain, recall, reflect, list_banks, create_bank (all with bank_id parameter)
-       - Bank ID from: X-Bank-Id header or HINDSIGHT_MCP_BANK_ID env var (default: "default")
-       - Example: claude mcp add --transport http hindsight http://localhost:8888/mcp --header "X-Bank-Id: my-bank"
-
-    Args:
-        memory: MemoryEngine instance
+    Returns the servers and apps separately so lifespans can be chained before
+    the middleware wraps the main app.
 
     Returns:
-        ASGI application
+        Tuple of (multi_bank_server, single_bank_server, multi_bank_app, single_bank_app)
     """
-    return MCPMiddleware(None, memory)
+    multi_bank_server = create_mcp_server(memory, multi_bank=True)
+    multi_bank_app = multi_bank_server.http_app(path="/")
+
+    single_bank_server = create_mcp_server(memory, multi_bank=False)
+    single_bank_app = single_bank_server.http_app(path="/")
+
+    return multi_bank_server, single_bank_server, multi_bank_app, single_bank_app

hindsight-api/hindsight_api/mcp_tools.py

@@ -552,6 +552,19 @@ def _register_create_bank(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsCo
             return f'{{"error": "{e}"}}'
 
 
+def _validate_mental_model_inputs(
+    name: str | None = None, source_query: str | None = None, max_tokens: int | None = None
+) -> str | None:
+    """Validate mental model inputs, returning an error message or None if valid."""
+    if name is not None and not name.strip():
+        return "name cannot be empty"
+    if source_query is not None and not source_query.strip():
+        return "source_query cannot be empty"
+    if max_tokens is not None and (max_tokens < 256 or max_tokens > 8192):
+        return f"max_tokens must be between 256 and 8192, got {max_tokens}"
+    return None
+
+
 # =========================================================================
 # MENTAL MODEL TOOLS
 # =========================================================================
@@ -656,7 +669,7 @@ def _register_get_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MCPTo
                     request_context=_get_request_context(config),
                 )
                 if model is None:
-                    return json.dumps({"error": f"Mental model '{mental_model_id}' not found"})
+                    return json.dumps({"error": f"Mental model '{mental_model_id}' not found in bank '{target_bank}'"})
                 return json.dumps(model, indent=2, default=str)
             except Exception as e:
                 logger.error(f"Error getting mental model: {e}", exc_info=True)
@@ -688,7 +701,7 @@ def _register_get_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MCPTo
                     request_context=_get_request_context(config),
                 )
                 if model is None:
-                    return {"error": f"Mental model '{mental_model_id}' not found"}
+                    return {"error": f"Mental model '{mental_model_id}' not found in bank '{target_bank}'"}
                 return model
             except Exception as e:
                 logger.error(f"Error getting mental model: {e}", exc_info=True)
@@ -734,6 +747,12 @@ def _register_create_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MC
                 if target_bank is None:
                     return '{"error": "No bank_id configured"}'
 
+                validation_error = _validate_mental_model_inputs(
+                    name=name, source_query=source_query, max_tokens=max_tokens
+                )
+                if validation_error:
+                    return json.dumps({"error": validation_error})
+
                 request_context = _get_request_context(config)
 
                 # Create with placeholder content
@@ -803,6 +822,12 @@ def _register_create_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MC
                 if target_bank is None:
                     return {"error": "No bank_id configured"}
 
+                validation_error = _validate_mental_model_inputs(
+                    name=name, source_query=source_query, max_tokens=max_tokens
+                )
+                if validation_error:
+                    return {"error": validation_error}
+
                 request_context = _get_request_context(config)
 
                 model = await memory.create_mental_model(
@@ -868,6 +893,12 @@ def _register_update_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MC
                 if target_bank is None:
                     return '{"error": "No bank_id configured"}'
 
+                validation_error = _validate_mental_model_inputs(
+                    name=name, source_query=source_query, max_tokens=max_tokens
+                )
+                if validation_error:
+                    return json.dumps({"error": validation_error})
+
                 model = await memory.update_mental_model(
                     bank_id=target_bank,
                     mental_model_id=mental_model_id,
@@ -878,7 +909,7 @@ def _register_update_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MC
                     request_context=_get_request_context(config),
                 )
                 if model is None:
-                    return json.dumps({"error": f"Mental model '{mental_model_id}' not found"})
+                    return json.dumps({"error": f"Mental model '{mental_model_id}' not found in bank '{target_bank}'"})
                 return json.dumps(model, indent=2, default=str)
             except Exception as e:
                 logger.error(f"Error updating mental model: {e}", exc_info=True)
@@ -912,6 +943,12 @@ def _register_update_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MC
                 if target_bank is None:
                     return {"error": "No bank_id configured"}
 
+                validation_error = _validate_mental_model_inputs(
+                    name=name, source_query=source_query, max_tokens=max_tokens
+                )
+                if validation_error:
+                    return {"error": validation_error}
+
                 model = await memory.update_mental_model(
                     bank_id=target_bank,
                     mental_model_id=mental_model_id,
@@ -922,7 +959,7 @@ def _register_update_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MC
                     request_context=_get_request_context(config),
                 )
                 if model is None:
-                    return {"error": f"Mental model '{mental_model_id}' not found"}
+                    return {"error": f"Mental model '{mental_model_id}' not found in bank '{target_bank}'"}
                 return model
             except Exception as e:
                 logger.error(f"Error updating mental model: {e}", exc_info=True)
@@ -959,7 +996,7 @@ def _register_delete_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MC
                     request_context=_get_request_context(config),
                 )
                 if not deleted:
-                    return json.dumps({"error": f"Mental model '{mental_model_id}' not found"})
+                    return json.dumps({"error": f"Mental model '{mental_model_id}' not found in bank '{target_bank}'"})
                 return json.dumps({"status": "deleted", "mental_model_id": mental_model_id})
             except Exception as e:
                 logger.error(f"Error deleting mental model: {e}", exc_info=True)
@@ -990,7 +1027,7 @@ def _register_delete_mental_model(mcp: FastMCP, memory: MemoryEngine, config: MC
                     request_context=_get_request_context(config),
                 )
                 if not deleted:
-                    return {"error": f"Mental model '{mental_model_id}' not found"}
+                    return {"error": f"Mental model '{mental_model_id}' not found in bank '{target_bank}'"}
                 return {"status": "deleted", "mental_model_id": mental_model_id}
             except Exception as e:
                 logger.error(f"Error deleting mental model: {e}", exc_info=True)

hindsight-api/tests/test_mcp_endpoint_routing.py

@@ -1,6 +1,7 @@
 """Integration test for MCP endpoint routing.
 
-This test verifies that /mcp/ and /mcp/{bank_id}/ expose different tool sets.
+This test verifies that /mcp/ and /mcp/{bank_id}/ expose different tool sets,
+and that URLs with or without trailing slashes both work (no 307 redirect).
 """
 
 import httpx
@@ -84,3 +85,196 @@ async def test_mcp_endpoint_routing_integration(memory):
                     assert retain_tool is not None
                     single_params = set(retain_tool.inputSchema.get("properties", {}).keys())
                     assert "bank_id" not in single_params, "Single-bank retain should NOT have bank_id parameter"
+
+
+@pytest.mark.asyncio
+async def test_mcp_no_trailing_slash_works(memory):
+    """Test that /mcp (no trailing slash) discovers tools without 307 redirect.
+
+    Starlette's Mount class redirects /mcp to /mcp/ with a 307 Temporary Redirect.
+    Many MCP clients don't follow POST redirects, causing 0 tools to be discovered.
+    MCPMiddleware wraps the app directly (no Mount), so the redirect never happens.
+    """
+    from hindsight_api.api import create_app
+
+    app = create_app(memory, mcp_api_enabled=True, initialize_memory=False)
+
+    async with app.router.lifespan_context(app):
+        from httpx import ASGITransport
+
+        async with httpx.AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as http_client:
+            # /mcp (no slash) should work the same as /mcp/
+            async with streamable_http_client("http://test/mcp", http_client=http_client) as (
+                read_stream,
+                write_stream,
+                _,
+            ):
+                async with ClientSession(read_stream, write_stream) as session:
+                    await session.initialize()
+                    result = await session.list_tools()
+
+                    tools = {t.name for t in result.tools}
+                    assert len(tools) >= 11, f"Expected at least 11 tools from /mcp, got {len(tools)}: {tools}"
+                    assert "retain" in tools
+                    assert "recall" in tools
+                    assert "list_banks" in tools
+
+            # /mcp/my-bank (single-bank, no slash) should also work
+            async with streamable_http_client("http://test/mcp/my-bank", http_client=http_client) as (
+                read_stream,
+                write_stream,
+                _,
+            ):
+                async with ClientSession(read_stream, write_stream) as session:
+                    await session.initialize()
+                    result = await session.list_tools()
+
+                    tools = {t.name for t in result.tools}
+                    assert "retain" in tools
+                    assert "list_banks" not in tools, "Single-bank /mcp/my-bank should NOT expose list_banks"
+
+
+@pytest.mark.asyncio
+async def test_mcp_tool_execution_through_client(memory):
+    """Test that tools can be called (not just discovered) through the MCP client.
+
+    This verifies the full pipeline: HTTP → middleware → FastMCP → tool → engine → response.
+    Previous tests only checked tool discovery (list_tools), not actual execution.
+    """
+    from httpx import ASGITransport
+
+    from hindsight_api.api import create_app
+
+    app = create_app(memory, mcp_api_enabled=True, initialize_memory=False)
+
+    async with app.router.lifespan_context(app):
+        async with httpx.AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as http_client:
+            async with streamable_http_client("http://test/mcp/", http_client=http_client) as (
+                read_stream,
+                write_stream,
+                _,
+            ):
+                async with ClientSession(read_stream, write_stream) as session:
+                    await session.initialize()
+
+                    # Execute list_banks tool
+                    result = await session.call_tool("list_banks", arguments={})
+                    assert result is not None
+                    assert len(result.content) > 0
+                    # The result text should be valid JSON with a "banks" key
+                    import json
+
+                    response_text = result.content[0].text
+                    parsed = json.loads(response_text)
+                    assert "banks" in parsed
+
+
+@pytest.mark.asyncio
+async def test_mcp_mental_model_validation_through_client(memory):
+    """Test that input validation works through the real MCP transport.
+
+    Verifies that invalid inputs return error messages without crashing,
+    and that the engine is never called with invalid data.
+    """
+    from httpx import ASGITransport
+
+    from hindsight_api.api import create_app
+
+    app = create_app(memory, mcp_api_enabled=True, initialize_memory=False)
+
+    async with app.router.lifespan_context(app):
+        async with httpx.AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as http_client:
+            async with streamable_http_client("http://test/mcp/", http_client=http_client) as (
+                read_stream,
+                write_stream,
+                _,
+            ):
+                async with ClientSession(read_stream, write_stream) as session:
+                    await session.initialize()
+
+                    # Test: empty name should return validation error
+                    import json
+
+                    result = await session.call_tool(
+                        "create_mental_model",
+                        arguments={"name": "", "source_query": "test query"},
+                    )
+                    assert result is not None
+                    parsed = json.loads(result.content[0].text)
+                    assert "error" in parsed
+                    assert "name cannot be empty" in parsed["error"]
+
+                    # Test: max_tokens out of range should return validation error
+                    result = await session.call_tool(
+                        "create_mental_model",
+                        arguments={"name": "Test", "source_query": "test query", "max_tokens": 0},
+                    )
+                    parsed = json.loads(result.content[0].text)
+                    assert "error" in parsed
+                    assert "max_tokens must be between 256 and 8192" in parsed["error"]
+
+
+@pytest.mark.asyncio
+async def test_mcp_bank_named_sse_routes_to_single_bank(memory):
+    """Test that a bank named 'sse' routes to single-bank mode.
+
+    Regression test: the old MCP_ENDPOINTS blocklist prevented banks named 'sse'
+    or 'messages' from being accessed via path routing. They fell through to
+    multi-bank mode instead.
+    """
+    from httpx import ASGITransport
+
+    from hindsight_api.api import create_app
+
+    app = create_app(memory, mcp_api_enabled=True, initialize_memory=False)
+
+    async with app.router.lifespan_context(app):
+        async with httpx.AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as http_client:
+            async with streamable_http_client("http://test/mcp/sse/", http_client=http_client) as (
+                read_stream,
+                write_stream,
+                _,
+            ):
+                async with ClientSession(read_stream, write_stream) as session:
+                    await session.initialize()
+                    result = await session.list_tools()
+                    tools = {t.name for t in result.tools}
+
+                    # Should be single-bank mode (no bank management tools)
+                    assert "retain" in tools
+                    assert "recall" in tools
+                    assert "list_banks" not in tools, "Bank 'sse' should route to single-bank mode"
+                    assert "create_bank" not in tools
+
+                    # retain should NOT have bank_id parameter (single-bank mode)
+                    retain_tool = next(t for t in result.tools if t.name == "retain")
+                    params = set(retain_tool.inputSchema.get("properties", {}).keys())
+                    assert "bank_id" not in params
+
+
+@pytest.mark.asyncio
+async def test_mcp_bank_named_messages_routes_to_single_bank(memory):
+    """Test that a bank named 'messages' routes to single-bank mode.
+
+    Same regression test as test_mcp_bank_named_sse_routes_to_single_bank but for 'messages'.
+    """
+    from httpx import ASGITransport
+
+    from hindsight_api.api import create_app
+
+    app = create_app(memory, mcp_api_enabled=True, initialize_memory=False)
+
+    async with app.router.lifespan_context(app):
+        async with httpx.AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as http_client:
+            async with streamable_http_client("http://test/mcp/messages/", http_client=http_client) as (
+                read_stream,
+                write_stream,
+                _,
+            ):
+                async with ClientSession(read_stream, write_stream) as session:
+                    await session.initialize()
+                    result = await session.list_tools()
+                    tools = {t.name for t in result.tools}
+
+                    assert "retain" in tools
+                    assert "list_banks" not in tools, "Bank 'messages' should route to single-bank mode"

hindsight-api/tests/test_mcp_extension.py

@@ -165,5 +165,5 @@ class TestMCPExtensionIntegration:
         assert "create_bank" in tools
         # Extension tool also present
         assert "test_extension_tool" in tools
-        # Total: 11 core + 1 extension = 12 tools
-        assert len(tools) == 12
+        # At least 11 core + 1 extension = 12 tools (may grow as new tools are added)
+        assert len(tools) >= 12

hindsight-api/tests/test_mcp_routing.py

@@ -354,9 +354,11 @@ async def test_middleware_handles_both_endpoints(mock_memory):
 
 @pytest.mark.asyncio
 async def test_routing_logic_from_url_path():
-    """Test that routing correctly selects server based on URL structure."""
-    from unittest.mock import AsyncMock
+    """Test that routing correctly selects server based on URL structure.
 
+    Simulates the path parsing logic from MCPMiddleware.__call__ after the
+    prefix has been stripped. Any first path segment is treated as a bank_id.
+    """
     from hindsight_api.api.mcp import MCPMiddleware
 
     # Mock memory
@@ -366,28 +368,23 @@ async def test_routing_logic_from_url_path():
     middleware = MCPMiddleware(None, mock_memory)
 
     # Simulate different URL patterns and verify routing
+    # Path is what remains after stripping the /mcp prefix
     test_cases = [
-        # (path_after_stripping_mcp, expected_bank_id_from_path, expected_bank_id, description)
+        # (path_after_prefix_strip, expected_bank_id_from_path, expected_bank_id, description)
         ("/alice/messages", True, "alice", "Bank ID in path with endpoint"),
         ("/my-agent-123/", True, "my-agent-123", "Bank ID in path with trailing slash"),
-        ("ciccio/messages", True, "ciccio", "Bank ID without leading slash (after mount strip)"),
-        ("bob", True, "bob", "Bank ID only, no leading slash"),
-        ("/messages", False, None, "MCP endpoint, no bank ID"),
+        ("/sse/", True, "sse", "Bank named 'sse' routes to single-bank"),
+        ("/messages/", True, "messages", "Bank named 'messages' routes to single-bank"),
         ("/", False, None, "Root path, no bank ID"),
     ]
 
     for path, expected_bank_from_path, expected_bank_id, description in test_cases:
-        # Simulate the path parsing logic with leading slash normalization
-        if path and not path.startswith("/"):
-            path = "/" + path
-
         bank_id = None
         bank_id_from_path = False
-        MCP_ENDPOINTS = {"sse", "messages"}
 
         if path.startswith("/") and len(path) > 1:
             parts = path[1:].split("/", 1)
-            if parts[0] and parts[0] not in MCP_ENDPOINTS:
+            if parts[0]:
                 bank_id = parts[0]
                 bank_id_from_path = True
 

hindsight-api/tests/test_mcp_tools.py

@@ -5,7 +5,13 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
-from hindsight_api.mcp_tools import MCPToolsConfig, build_content_dict, parse_timestamp, register_mcp_tools
+from hindsight_api.mcp_tools import (
+    MCPToolsConfig,
+    _validate_mental_model_inputs,
+    build_content_dict,
+    parse_timestamp,
+    register_mcp_tools,
+)
 
 
 class TestParseTimestamp:
@@ -546,3 +552,95 @@ class TestRefreshMentalModel:
         mock_memory.submit_async_refresh_mental_model.side_effect = RuntimeError("DB error")
         result = await _tools(mcp_server_with_mental_models)["refresh_mental_model"].fn(mental_model_id="mm-1")
         assert "error" in result
+
+
+class TestValidateMentalModelInputs:
+    """Tests for the _validate_mental_model_inputs helper."""
+
+    def test_valid_inputs(self):
+        assert _validate_mental_model_inputs(name="Test", source_query="query", max_tokens=2048) is None
+
+    def test_none_inputs(self):
+        assert _validate_mental_model_inputs() is None
+
+    def test_empty_name(self):
+        result = _validate_mental_model_inputs(name="")
+        assert result == "name cannot be empty"
+
+    def test_whitespace_name(self):
+        result = _validate_mental_model_inputs(name="   ")
+        assert result == "name cannot be empty"
+
+    def test_empty_source_query(self):
+        result = _validate_mental_model_inputs(source_query="")
+        assert result == "source_query cannot be empty"
+
+    def test_whitespace_source_query(self):
+        result = _validate_mental_model_inputs(source_query="  \t  ")
+        assert result == "source_query cannot be empty"
+
+    def test_max_tokens_too_low(self):
+        result = _validate_mental_model_inputs(max_tokens=0)
+        assert "max_tokens must be between 256 and 8192" in result
+
+    def test_max_tokens_too_high(self):
+        result = _validate_mental_model_inputs(max_tokens=10000)
+        assert "max_tokens must be between 256 and 8192" in result
+
+    def test_max_tokens_at_lower_bound(self):
+        assert _validate_mental_model_inputs(max_tokens=256) is None
+
+    def test_max_tokens_at_upper_bound(self):
+        assert _validate_mental_model_inputs(max_tokens=8192) is None
+
+
+@pytest.mark.asyncio
+class TestMentalModelInputValidation:
+    """Tests that validation is applied in create/update tools before engine calls."""
+
+    async def test_create_empty_name_returns_error_multi_bank(self, mcp_server_with_mental_models, mock_memory):
+        result = await _tools(mcp_server_with_mental_models)["create_mental_model"].fn(name="", source_query="query")
+        assert "name cannot be empty" in result
+        mock_memory.create_mental_model.assert_not_called()
+
+    async def test_create_empty_source_query_returns_error_multi_bank(self, mcp_server_with_mental_models, mock_memory):
+        result = await _tools(mcp_server_with_mental_models)["create_mental_model"].fn(name="Test", source_query="")
+        assert "source_query cannot be empty" in result
+        mock_memory.create_mental_model.assert_not_called()
+
+    async def test_create_max_tokens_too_low_multi_bank(self, mcp_server_with_mental_models, mock_memory):
+        result = await _tools(mcp_server_with_mental_models)["create_mental_model"].fn(
+            name="Test", source_query="query", max_tokens=0
+        )
+        assert "max_tokens must be between 256 and 8192" in result
+        mock_memory.create_mental_model.assert_not_called()
+
+    async def test_create_max_tokens_too_high_single_bank(self, mcp_server_single_bank, mock_memory):
+        result = await _tools(mcp_server_single_bank)["create_mental_model"].fn(
+            name="Test", source_query="query", max_tokens=10000
+        )
+        assert isinstance(result, dict)
+        assert "max_tokens must be between 256 and 8192" in result["error"]
+        mock_memory.create_mental_model.assert_not_called()
+
+    async def test_update_empty_name_returns_error_multi_bank(self, mcp_server_with_mental_models, mock_memory):
+        result = await _tools(mcp_server_with_mental_models)["update_mental_model"].fn(mental_model_id="mm-1", name="")
+        assert "name cannot be empty" in result
+        mock_memory.update_mental_model.assert_not_called()
+
+    async def test_update_empty_name_returns_error_single_bank(self, mcp_server_single_bank, mock_memory):
+        result = await _tools(mcp_server_single_bank)["update_mental_model"].fn(mental_model_id="mm-1", name="  ")
+        assert isinstance(result, dict)
+        assert "name cannot be empty" in result["error"]
+        mock_memory.update_mental_model.assert_not_called()
+
+    async def test_not_found_error_includes_bank_id_multi_bank(self, mcp_server_with_mental_models, mock_memory):
+        mock_memory.get_mental_model.return_value = None
+        result = await _tools(mcp_server_with_mental_models)["get_mental_model"].fn(mental_model_id="missing")
+        assert "test-bank" in result
+
+    async def test_not_found_error_includes_bank_id_single_bank(self, mcp_server_single_bank, mock_memory):
+        mock_memory.get_mental_model.return_value = None
+        result = await _tools(mcp_server_single_bank)["get_mental_model"].fn(mental_model_id="missing")
+        assert isinstance(result, dict)
+        assert "fixed-bank" in result["error"]