Add low-level MCP contrib package with tools and tests

AGENTS.md

@@ -0,0 +1,41 @@
+# AGENTS.md
+
+## Project Overview
+- Purpose: Guillotina core framework and official contrib addons.
+- Main stack: Python async API server (ASGI), PostgreSQL, optional Redis.
+- Key paths:
+  - `guillotina/` core framework and contrib packages
+  - `guillotina/tests/` test suite
+  - `docs/source/` documentation
+
+## Development Commands
+- Setup (local venv expected at repo root):
+  - `python3 -m venv .venv`
+  - `source .venv/bin/activate`
+  - `pip install -r requirements.txt`
+  - `pip install -r contrib-requirements.txt`
+  - `pip install -e '.[test]'`
+- Run local server:
+  - `g` (uses `config.yaml` by default)
+- Run tests:
+  - `.venv/bin/pytest guillotina/tests`
+  - Targeted: `.venv/bin/pytest guillotina/tests/<path>`
+
+## Validation
+- For contrib changes, run focused tests under the touched contrib test folder.
+- For API/service changes, verify status codes and response payload contracts.
+- Keep docs updated under `docs/source/contrib/` when adding contrib features.
+
+## Deployment Notes
+- This repo is a framework/library; no direct client deployment from this repo by default.
+- Build/release lifecycle should follow package versioning (`VERSION`, `CHANGELOG.rst`).
+
+## Constraints / Gotchas
+- Keep compatibility with repository formatting (`black` line length 110).
+- Avoid wrapper layers when task explicitly requires low-level protocol primitives.
+- Never commit credentials or local environment files.
+
+## Task Closeout Notes
+- Update `CHANGELOG.rst` for notable changes.
+- Record branch name, commit hash, validation output, and task evidence in Ops Tracker.
+

CHANGELOG.rst

@@ -4,6 +4,11 @@ CHANGELOG
 7.0.7 (unreleased)
 ------------------
 
+- Add `guillotina.contrib.mcp` with low-level MCP server integration
+  (`mcp.server.lowlevel`), tool registry utility, MCP services,
+  cache invalidation subscribers, and tests/docs coverage.
+- Optimize MCP `list_children` tool to prefer catalog queries and
+  fallback to `async_items` when catalog is unavailable.
 - Docs: Update documentation and configuration settings
 - Chore: Update sphinx-guillotina-theme version to 1.0.9
   [rboixaderg]

contrib-requirements.txt

@@ -13,6 +13,7 @@ MarkupSafe<2.1.0
 pytz==2020.1
 emcache==0.6.0; python_version < '3.10'
 pymemcache==3.4.0; python_version < '3.10'
+mcp>=1.0.0; python_version >= '3.10'
 
 # Conditional Pillow versions
 pillow==10.4.0; python_version < '3.11'

docs/source/contrib/index.rst

@@ -17,3 +17,4 @@ Contents:
    swagger
    mailer
    dbusers
+   mcp

docs/source/contrib/mcp.md

@@ -0,0 +1,44 @@
+# MCP
+
+`guillotina.contrib.mcp` provides a low-level MCP integration layer built on
+the official `mcp.server.lowlevel` primitives (no FastMCP wrapper).
+
+## Installation
+
+```bash
+pip install "guillotina[mcp]"
+```
+
+## Configuration
+
+```yaml
+applications:
+  - guillotina
+  - guillotina.contrib.mcp
+
+mcp:
+  enabled: true
+  server_name: guillotina-mcp
+  default_child_limit: 50
+```
+
+## Runtime endpoints
+
+- `GET /@mcp`: registry metadata and registered tools.
+- `GET /@mcp/tools`: tool list and schemas.
+- `POST /@mcp/tools/invoke`: executes one tool with payload
+  `{ "tool": "<name>", "arguments": { ... } }`.
+- `GET /@mcp/server/status`: validates low-level SDK availability.
+
+## Built-in tools
+
+- `resolve_path`: resolve a path and return basic metadata.
+- `list_children`: list child resources from a folder-like resource.
+- `serialize_resource`: execute Guillotina serialization adapters.
+- `notify_modified`: emit an `ObjectModifiedEvent`.
+
+The tool registry is implemented as a Guillotina utility and cache invalidation
+is handled by subscribers on object add/modify/remove events.
+
+`list_children` prefers catalog-backed lookup when a catalog utility is
+available and falls back to `async_items()` iteration when it is not.

guillotina/contrib/mcp/__init__.py

@@ -0,0 +1,24 @@
+from guillotina import configure
+
+
+app_settings = {
+    "mcp": {
+        "enabled": True,
+        "server_name": "guillotina-mcp",
+        "default_child_limit": 50,
+    },
+    "load_utilities": {
+        "mcp_tool_registry": {
+            "provides": "guillotina.contrib.mcp.interfaces.IMCPToolRegistry",
+            "factory": "guillotina.contrib.mcp.backend.MCPToolRegistry",
+            "settings": {},
+        }
+    },
+}
+
+
+def includeme(root, settings):
+    configure.scan("guillotina.contrib.mcp.install")
+    configure.scan("guillotina.contrib.mcp.permissions")
+    configure.scan("guillotina.contrib.mcp.services")
+    configure.scan("guillotina.contrib.mcp.subscribers")

guillotina/contrib/mcp/backend.py

@@ -0,0 +1,195 @@
+from dataclasses import dataclass
+from guillotina import app_settings
+from guillotina.contrib.mcp import resources as mcp_resources
+from guillotina.contrib.mcp import tools
+from typing import Any
+from typing import Awaitable
+from typing import Callable
+from typing import Dict
+from typing import List
+from typing import Optional
+
+import json
+
+
+ToolHandler = Callable[[Any, Any, Dict[str, Any]], Awaitable[Dict[str, Any]]]
+ResourceHandler = Callable[[Any], Awaitable[Dict[str, Any]]]
+
+
+@dataclass
+class MCPTool:
+    name: str
+    description: str
+    input_schema: Dict[str, Any]
+    handler: ToolHandler
+    cacheable: bool = False
+
+
+@dataclass
+class MCPResource:
+    name: str
+    uri: str
+    description: str
+    endpoint: str
+    handler: ResourceHandler
+    mime_type: str = "application/json"
+
+
+class MCPToolRegistry:
+    def __init__(self, settings: Optional[Dict[str, Any]] = None):
+        config = app_settings.get("mcp", {})
+        self._settings = settings or {}
+        self._enabled = bool(config.get("enabled", True))
+        self._server_name = str(config.get("server_name", "guillotina-mcp"))
+        self._default_child_limit = int(config.get("default_child_limit", 50))
+        self._tools: Dict[str, MCPTool] = {}
+        self._resources: Dict[str, MCPResource] = {}
+        self._cache: Dict[str, Dict[str, Any]] = {}
+        self._cache_revision = 0
+        self._register_default_tools()
+        self._register_default_resources()
+
+    def _register_default_tools(self) -> None:
+        for tool_name, description, input_schema, handler, cacheable in tools.default_tools(
+            self._default_child_limit
+        ):
+            self.register_tool(
+                name=tool_name,
+                description=description,
+                input_schema=input_schema,
+                handler=handler,
+                cacheable=cacheable,
+            )
+
+    def _register_default_resources(self) -> None:
+        for name, uri, description, endpoint, handler in mcp_resources.default_resources():
+            self.register_resource(
+                name=name,
+                uri=uri,
+                description=description,
+                endpoint=endpoint,
+                handler=handler,
+            )
+
+    def is_enabled(self) -> bool:
+        return self._enabled
+
+    def register_tool(
+        self,
+        *,
+        name: str,
+        description: str,
+        input_schema: Dict[str, Any],
+        handler: ToolHandler,
+        cacheable: bool = False,
+    ) -> None:
+        clean_name = str(name or "").strip()
+        if not clean_name:
+            raise ValueError("Tool name is required")
+        self._tools[clean_name] = MCPTool(
+            name=clean_name,
+            description=str(description or "").strip(),
+            input_schema=input_schema or {"type": "object"},
+            handler=handler,
+            cacheable=cacheable,
+        )
+
+    def list_tools(self) -> List[Dict[str, Any]]:
+        return [
+            {
+                "name": tool.name,
+                "description": tool.description,
+                "inputSchema": tool.input_schema,
+                "cacheable": tool.cacheable,
+            }
+            for tool in sorted(self._tools.values(), key=lambda registered: registered.name)
+        ]
+
+    # ── Resource management ──────────────────────────────────────────
+
+    def register_resource(
+        self,
+        *,
+        name: str,
+        uri: str,
+        description: str,
+        endpoint: str,
+        handler: ResourceHandler,
+        mime_type: str = "application/json",
+    ) -> None:
+        clean_name = str(name or "").strip()
+        if not clean_name:
+            raise ValueError("Resource name is required")
+        self._resources[clean_name] = MCPResource(
+            name=clean_name,
+            uri=str(uri or "").strip(),
+            description=str(description or "").strip(),
+            endpoint=str(endpoint or "").strip(),
+            handler=handler,
+            mime_type=mime_type,
+        )
+
+    def list_resources(self) -> List[Dict[str, Any]]:
+        return [
+            {
+                "uri": res.uri,
+                "name": res.name,
+                "description": res.description,
+                "endpoint": res.endpoint,
+                "mimeType": res.mime_type,
+            }
+            for res in sorted(self._resources.values(), key=lambda r: r.name)
+        ]
+
+    async def read_resource(self, resource_name: str, context: Any, request: Any) -> Dict[str, Any]:
+        clean_name = str(resource_name or "").strip()
+        if clean_name not in self._resources:
+            raise ValueError(f"Unknown MCP resource: {resource_name}")
+        resource = self._resources[clean_name]
+        return await resource.handler(request)
+
+    def _cache_key(self, tool_name: str, arguments: Dict[str, Any]) -> str:
+        payload = json.dumps(arguments, sort_keys=True, separators=(",", ":"), default=str)
+        return f"{tool_name}:{payload}"
+
+    async def invoke(
+        self, tool_name: str, context: Any, request: Any, arguments: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
+        clean_name = str(tool_name or "").strip()
+        if clean_name not in self._tools:
+            raise ValueError(f"Unknown MCP tool: {tool_name}")
+
+        clean_arguments = arguments or {}
+        if not isinstance(clean_arguments, dict):
+            raise ValueError("Tool arguments must be an object")
+
+        tool = self._tools[clean_name]
+        cache_key = self._cache_key(clean_name, clean_arguments)
+        if tool.cacheable and cache_key in self._cache:
+            return self._cache[cache_key]
+
+        result = await tool.handler(context, request, clean_arguments)
+        if tool.cacheable:
+            self._cache[cache_key] = result
+        return result
+
+    def invalidate_cache(self, reason: str = "manual") -> None:
+        self._cache.clear()
+        self._cache_revision += 1
+
+    def metadata(self) -> Dict[str, Any]:
+        return {
+            "enabled": self.is_enabled(),
+            "server_name": self._server_name,
+            "tool_count": len(self._tools),
+            "resource_count": len(self._resources),
+            "cache_revision": self._cache_revision,
+        }
+
+    def create_lowlevel_server(self, context: Any = None, request: Any = None) -> Any:
+        from guillotina.contrib.mcp.server import LowLevelMCPServer
+
+        adapter = LowLevelMCPServer(
+            registry=self, context=context, request=request, server_name=self._server_name
+        )
+        return adapter.build()

guillotina/contrib/mcp/install.py

@@ -0,0 +1,16 @@
+from guillotina import configure
+from guillotina.addons import Addon
+from guillotina.contrib.mcp.interfaces import IMCPSettings
+from guillotina.utils import get_registry
+
+
+@configure.addon(name="mcp", title="Guillotina MCP integration")
+class MCPAddon(Addon):
+    @classmethod
+    async def install(cls, container, request):
+        registry = await get_registry()
+        registry.register_interface(IMCPSettings)
+
+    @classmethod
+    async def uninstall(cls, container, request):
+        pass

guillotina/contrib/mcp/interfaces.py

@@ -0,0 +1,31 @@
+from guillotina import schema
+from zope.interface import Interface
+
+
+class IMCPSettings(Interface):
+    enabled = schema.Bool(title="Enable MCP services", default=True, required=False)
+    server_name = schema.TextLine(title="Low-level MCP server name", default="guillotina-mcp", required=False)
+    default_child_limit = schema.Int(title="Default list_children limit", default=50, required=False)
+
+
+class IMCPToolRegistry(Interface):
+    def list_tools():
+        """Return registered MCP tools."""
+
+    def list_resources():
+        """Return registered MCP resources."""
+
+    async def invoke(tool_name, context, request, arguments=None):
+        """Execute one tool and return a JSON-serializable response."""
+
+    async def read_resource(resource_name, context, request):
+        """Read one resource and return a JSON-serializable response."""
+
+    def metadata():
+        """Return metadata for diagnostics."""
+
+    def invalidate_cache(reason="manual"):
+        """Invalidate cached tool responses."""
+
+    def create_lowlevel_server(context=None, request=None):
+        """Build a low-level MCP server object."""

guillotina/contrib/mcp/permissions.py

@@ -0,0 +1,9 @@
+from guillotina import configure
+
+
+configure.permission("guillotina.MCPView", "View MCP integration services")
+configure.permission("guillotina.MCPExecute", "Execute MCP tools")
+
+configure.grant(permission="guillotina.MCPView", role="guillotina.Manager")
+configure.grant(permission="guillotina.MCPView", role="guillotina.Owner")
+configure.grant(permission="guillotina.MCPExecute", role="guillotina.Manager")