Robert Helewka
5527cf6bdb
Workspace scoping is the integration's security-critical property: an agent in workspace A must never see content from workspace B or from any global library, regardless of what the calling LLM tries. Adds `workspace_id` to SearchRequest with __post_init__ normalization that converts empty strings to None — so "" cannot slip through as a truthy filter at the Cypher boundary. Extracts the workspace scope clause to a single string and appends it to all five search queries (vector, fulltext-chunk, fulltext-concept, graph, image): ($workspace_id IS NULL AND lib.workspace_id IS NULL OR lib.workspace_id = $workspace_id) Either workspace-only or global-only — never both — and the operator precedence is bracketed so a refactor can't accidentally widen it. A test verifies the literal clause string for that exact reason. Adds `workspace_id` as a parameter to every MCP tool (`search`, `get_chunk`, `list_libraries`, `list_collections`, `list_items`). Deliberately undocumented in tool docstrings so the calling LLM is never told the parameter exists — it is system-injected by Daedalus's chat path and force-overwritten before reaching Mnemosyne. Mnemosyne also validates the value but the security guarantee is enforced upstream. Adds the `get_health` MCP tool per the Pallas health spec: returns ok / degraded / error after probing Neo4j, S3, and the embedding model registration. Used by Daedalus's existing health poller. Updates the server INSTRUCTIONS string to advertise the new tool and the two new library types (business, finance). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
33 lines
838 B
Python
33 lines
838 B
Python
"""Tests that the FastMCP server registers the expected tools."""
|
|
|
|
import asyncio
|
|
|
|
from django.test import TestCase
|
|
|
|
from mcp_server.server import mcp
|
|
|
|
|
|
EXPECTED_TOOLS = {
|
|
"search",
|
|
"get_chunk",
|
|
"list_libraries",
|
|
"list_collections",
|
|
"list_items",
|
|
"get_health",
|
|
}
|
|
|
|
|
|
class ServerRegistrationTest(TestCase):
|
|
def test_expected_tools_registered(self):
|
|
tools = asyncio.run(mcp.get_tools())
|
|
self.assertEqual(EXPECTED_TOOLS, set(tools.keys()))
|
|
|
|
def test_tool_descriptions_within_limit(self):
|
|
tools = asyncio.run(mcp.get_tools())
|
|
for name, tool in tools.items():
|
|
description = tool.description or ""
|
|
self.assertLessEqual(
|
|
len(description), 1024,
|
|
f"Tool '{name}' description exceeds 1024 chars (MCP spec limit).",
|
|
)
|