MCP

Makefile

@@ -41,3 +41,12 @@ create-cockroachdb: ## Create CockroachDB
 	@echo ""
 	@echo "$(YELLOW)==> Creating CockroachDB $(VERSION)$(RESET)"
 	./bin/python _cockroachdb-createdb.py
+
+.PHONY: format
+format: ## Format code
+	flake8 guillotina --config=setup.cfg
+	black guillotina/
+	isort -rc guillotina/
+
+tests:
+	DATABASE=POSTGRES pytest -s -x guillotina

guillotina/contrib/mcp/__init__.py

@@ -0,0 +1,17 @@
+from guillotina import configure
+
+
+app_settings = {
+    "mcp": {
+        "enabled": True,
+        "description_extras": {},
+        "extra_tools_module": None,
+        "token_max_duration_days": 90,
+        "token_allowed_durations": None,
+    },
+}
+
+
+def includeme(root, settings):
+    configure.scan("guillotina.contrib.mcp.permissions")
+    configure.scan("guillotina.contrib.mcp.services")

guillotina/contrib/mcp/backend.py

@@ -0,0 +1,126 @@
+from contextvars import ContextVar
+from guillotina.component import get_multi_adapter
+from guillotina.component import query_utility
+from guillotina.interfaces import ICatalogUtility
+from guillotina.interfaces import IResource
+from guillotina.interfaces import IResourceSerializeToJson
+from guillotina.utils import get_object_by_uid
+from guillotina.utils import navigate_to
+
+import typing
+
+
+_mcp_context_var: ContextVar[typing.Optional[IResource]] = ContextVar("mcp_context", default=None)
+
+
+def get_mcp_context():
+    return _mcp_context_var.get()
+
+
+def set_mcp_context(context: IResource):
+    _mcp_context_var.set(context)
+
+
+def clear_mcp_context():
+    try:
+        _mcp_context_var.set(None)
+    except LookupError:
+        pass
+
+
+class InProcessBackend:
+    def _get_base_context(self) -> IResource:
+        ctx = get_mcp_context()
+        if ctx is None:
+            raise RuntimeError("MCP context not set (not in @mcp request?)")
+        return ctx
+
+    def _resolve_context(self, context: typing.Optional[IResource]) -> IResource:
+        if context is None:
+            return self._get_base_context()
+        if not IResource.providedBy(context):
+            raise RuntimeError(
+                "InProcessBackend requires IResource context. Use the @mcp endpoint or set MCP context."
+            )
+        return context
+
+    async def search(self, context: IResource, query: dict) -> dict:
+        base = self._resolve_context(context)
+        search = query_utility(ICatalogUtility)
+        if search is None:
+            return {"items": [], "items_total": 0}
+        return await search.search(base, query)
+
+    async def count(self, context: IResource, query: dict) -> int:
+        base = self._resolve_context(context)
+        search = query_utility(ICatalogUtility)
+        if search is None:
+            return 0
+        return await search.count(base, query)
+
+    async def get_content(
+        self,
+        context: IResource,
+        path: typing.Optional[str],
+        uid: typing.Optional[str],
+    ) -> dict:
+        from guillotina import task_vars
+
+        base = self._resolve_context(context)
+        request = task_vars.request.get()
+        if uid:
+            try:
+                ob = await get_object_by_uid(uid)
+            except KeyError:
+                return {}
+            if not self._in_container_tree(ob, base):
+                return {}
+        elif path is not None:
+            rel_path = path.strip("/") or ""
+            try:
+                ob = await navigate_to(base, "/" + rel_path) if rel_path else base
+            except KeyError:
+                return {}
+        else:
+            return {}
+        serializer = get_multi_adapter((ob, request), IResourceSerializeToJson)
+        return await serializer()
+
+    def _in_container_tree(self, ob: IResource, container: IResource) -> bool:
+        from guillotina.utils import get_content_path
+
+        ob_path = get_content_path(ob)
+        cont_path = get_content_path(container)
+        return ob_path == cont_path or ob_path.startswith(cont_path.rstrip("/") + "/")
+
+    async def list_children(
+        self,
+        context: IResource,
+        path: str,
+        _from: int = 0,
+        _size: int = 20,
+    ) -> dict:
+        base = self._resolve_context(context)
+        path = path.strip("/") or ""
+        try:
+            container = await navigate_to(base, "/" + path) if path else base
+        except KeyError:
+            return {"items": [], "items_total": 0}
+        from guillotina import task_vars
+        from guillotina.interfaces import IFolder
+        from guillotina.interfaces import IResourceSerializeToJsonSummary
+
+        if not IFolder.providedBy(container):
+            return {"items": [], "items_total": 0}
+        request = task_vars.request.get()
+        items = []
+        total = 0
+        async for name, child in container.async_items():
+            if total >= _from + _size:
+                total += 1
+                continue
+            if total >= _from:
+                summary_serializer = get_multi_adapter((child, request), IResourceSerializeToJsonSummary)
+                items.append(await summary_serializer())
+            total += 1
+        return {"items": items, "items_total": total}

guillotina/contrib/mcp/interfaces.py

@@ -0,0 +1,8 @@
+from zope.interface import Interface
+
+
+class IMCPDescriptionExtras(Interface):
+    """Utility returning a dict mapping tool name to extra description text
+    (appended to the base tool description for LLM context).
+    Tool names: search, count, get_content, list_children.
+    """

guillotina/contrib/mcp/permissions.py

@@ -0,0 +1,7 @@
+from guillotina import configure
+
+
+configure.permission("guillotina.mcp.Use", "Use MCP tools to query content")
+configure.grant(permission="guillotina.mcp.Use", role="guillotina.Authenticated")
+configure.permission("guillotina.mcp.IssueToken", "Issue a long-lived MCP token")
+configure.grant(permission="guillotina.mcp.IssueToken", role="guillotina.Authenticated")

guillotina/contrib/mcp/server.py

@@ -0,0 +1,31 @@
+from guillotina._settings import app_settings
+from guillotina.contrib.mcp.backend import InProcessBackend
+from guillotina.contrib.mcp.tools import register_tools
+
+
+_mcp_server = None
+_mcp_app = None
+
+
+def get_mcp_app_and_server():
+    global _mcp_server, _mcp_app
+    if _mcp_app is not None:
+        return _mcp_app, _mcp_server
+    from mcp.server.fastmcp import FastMCP
+
+    backend = InProcessBackend()
+    mcp = FastMCP(
+        "Guillotina MCP",
+        json_response=True,
+        stateless_http=True,
+    )
+    if hasattr(mcp, "settings") and hasattr(mcp.settings, "streamable_http_path"):
+        mcp.settings.streamable_http_path = "/"
+    register_tools(mcp, backend)
+    extra_module = app_settings.get("mcp", {}).get("extra_tools_module")
+    if extra_module:
+        mod = __import__(str(extra_module), fromlist=["register_extra_tools"])
+        getattr(mod, "register_extra_tools")(mcp, backend)
+    _mcp_server = mcp
+    _mcp_app = mcp.streamable_http_app()
+    return _mcp_app, _mcp_server

guillotina/contrib/mcp/services.py

@@ -0,0 +1,120 @@
+from guillotina import configure
+from guillotina._settings import app_settings
+from guillotina.api.service import Service
+from guillotina.auth import authenticate_user
+from guillotina.auth.users import AnonymousUser
+from guillotina.contrib.mcp.backend import clear_mcp_context
+from guillotina.contrib.mcp.backend import set_mcp_context
+from guillotina.contrib.mcp.server import get_mcp_app_and_server
+from guillotina.interfaces import IResource
+from guillotina.response import HTTPPreconditionFailed
+from guillotina.response import HTTPUnauthorized
+from guillotina.response import Response
+from guillotina.utils import get_authenticated_user
+
+import anyio
+import copy
+import logging
+
+
+logger = logging.getLogger("guillotina")
+
+
+@configure.service(
+    context=IResource,
+    method="POST",
+    permission="guillotina.mcp.Use",
+    name="@mcp",
+    summary="MCP protocol endpoint (POST)",
+)
+@configure.service(
+    context=IResource,
+    method="GET",
+    permission="guillotina.mcp.Use",
+    name="@mcp",
+    summary="MCP protocol endpoint (GET)",
+)
+async def mcp_service(context, request):
+    if not app_settings.get("mcp", {}).get("enabled", True):
+        from guillotina.response import HTTPNotFound
+
+        raise HTTPNotFound(content={"reason": "MCP is disabled"})
+    set_mcp_context(context)
+    try:
+        scope = copy.copy(request.scope)
+        scope["path"] = "/"
+        scope["raw_path"] = b"/"
+        app, server = get_mcp_app_and_server()
+        session_manager = server.session_manager
+        original_task_group = session_manager._task_group
+        async with anyio.create_task_group() as tg:
+            session_manager._task_group = tg
+            try:
+                await app(scope, request.receive, request.send)
+            finally:
+                session_manager._task_group = original_task_group
+    finally:
+        clear_mcp_context()
+    resp = Response()
+    resp._prepared = True
+    resp._eof_sent = True
+    return resp
+
+
+@configure.service(
+    context=IResource,
+    method="POST",
+    permission="guillotina.mcp.IssueToken",
+    name="@mcp-token",
+    summary="Issue a long-lived JWT for MCP client configuration",
+)
+class MCPToken(Service):
+    __body_required__ = False
+
+    async def __call__(self):
+        if not app_settings.get("mcp", {}).get("enabled", True):
+            raise HTTPPreconditionFailed(content={"reason": "MCP is disabled"})
+        user = get_authenticated_user()
+        if user is None or isinstance(user, AnonymousUser):
+            raise HTTPUnauthorized(content={"reason": "Authentication required"})
+        try:
+            body = await self.request.json()
+        except Exception:
+            body = {}
+        if not isinstance(body, dict):
+            body = {}
+        duration_days = body.get("duration_days", 30)
+        try:
+            duration_days = int(duration_days)
+        except (TypeError, ValueError):
+            raise HTTPPreconditionFailed(
+                content={"reason": "duration_days must be an integer", "value": duration_days}
+            )
+        mcp_settings = app_settings.get("mcp", {})
+        max_days = mcp_settings.get("token_max_duration_days", 90)
+        allowed = mcp_settings.get("token_allowed_durations")
+        if allowed is not None:
+            if duration_days not in allowed:
+                raise HTTPPreconditionFailed(
+                    content={
+                        "reason": "duration_days must be one of",
+                        "allowed": allowed,
+                        "value": duration_days,
+                    }
+                )
+        else:
+            if duration_days < 1 or duration_days > max_days:
+                raise HTTPPreconditionFailed(
+                    content={
+                        "reason": "duration_days must be between 1 and token_max_duration_days",
+                        "token_max_duration_days": max_days,
+                        "value": duration_days,
+                    }
+                )
+        timeout = duration_days * 24 * 3600
+        jwt_token, data = authenticate_user(user.id, data={"purpose": "mcp"}, timeout=timeout)
+        return {
+            "token": jwt_token,
+            "exp": data["exp"],
+            "duration_days": duration_days,
+        }