MCP: Add subsystem providing a few server and client utilities

Through the `ctk query mcp {list,inquire,launch}` subcommands.

Author: amotl

.github/workflows/main.yml

@@ -68,6 +68,9 @@ jobs:
     - name: Set up project
       run: |
 
+        # Install sponge.
+        sudo apt-get install moreutils
+
         # `setuptools 64` adds support for editable install hooks (PEP 660).
         # https://github.com/pypa/setuptools/blob/main/CHANGES.rst#v6400
         # `setuptools 71` significantly re-vendors setuptools packages, causing hiccups on Python 3.8.

CHANGES.md

@@ -1,6 +1,8 @@
 # Changelog
 
 ## Unreleased
+- MCP: Add subsystem providing a few server and client utilities through
+  the `ctk query mcp {list,inquire,launch}` subcommands.
 
 ## 2025/01/31 v0.0.31
 - Fixed connectivity for `jobstats collect`

cratedb_toolkit/query/cli.py

@@ -5,6 +5,7 @@
 
 from ..util.cli import boot_click
 from .convert.cli import convert_query
+from .mcp.cli import cli as mcp_cli
 
 logger = logging.getLogger(__name__)
 
@@ -16,9 +17,10 @@
 @click.pass_context
 def cli(ctx: click.Context, verbose: bool, debug: bool):
     """
-    Query expression utilities.
+    Query utilities.
     """
     return boot_click(ctx, verbose, debug)
 
 
 cli.add_command(convert_query, name="convert")
+cli.add_command(mcp_cli, name="mcp")

cratedb_toolkit/query/mcp/__init__.py

@@ -0,0 +1,92 @@
+import asyncio
+import logging
+
+import click
+from click_aliases import ClickAliasedGroup
+
+from cratedb_toolkit.query.mcp.inquiry import McpServerInquiry
+from cratedb_toolkit.query.mcp.registry import McpServerRegistry
+from cratedb_toolkit.query.mcp.util import format_output
+from cratedb_toolkit.util.cli import boot_click
+
+logger = logging.getLogger(__name__)
+
+
+def get_format_option(default="markdown"):
+    return click.option(
+        "--format",
+        "format_",
+        type=click.Choice(["markdown", "json", "yaml"]),
+        required=True,
+        default=default,
+        help="Select output format",
+    )
+
+
+format_option_markdown = get_format_option(default="markdown")
+format_option_json = get_format_option(default="json")
+
+
+@click.group(cls=ClickAliasedGroup)  # type: ignore[arg-type]
+@click.option("--server-name", type=str, required=False, help="Select MCP server name")
+@click.option("--verbose", is_flag=True, required=False, help="Turn on logging")
+@click.option("--debug", is_flag=True, required=False, help="Turn on logging with debug level")
+@click.version_option()
+@click.pass_context
+def cli(ctx: click.Context, server_name: str, verbose: bool, debug: bool):
+    """
+    MCP utilities.
+    """
+    ctx.meta["registry"] = McpServerRegistry()
+    ctx.meta["server_name"] = server_name
+    return boot_click(ctx, verbose, debug)
+
+
+@cli.command(name="list")
+@format_option_json
+@click.pass_context
+def list_servers(
+    ctx: click.Context,
+    format_: str,
+):
+    """
+    Enumerate registered MCP servers.
+    """
+    registry = ctx.meta["registry"]
+    server_name = ctx.meta["server_name"]
+    inquiry = McpServerInquiry(registry.select(server_name))
+    names = [server.name for server in inquiry.servers]
+    print(format_output(names, format_))  # noqa: T201
+
+
+@cli.command()
+@format_option_markdown
+@click.pass_context
+def inquire(
+    ctx: click.Context,
+    format_: str,
+):
+    """
+    Inquire MCP servers, report about their capabilities.
+    """
+    registry = ctx.meta["registry"]
+    server_name = ctx.meta["server_name"]
+    inquiry = McpServerInquiry(registry.select(server_name))
+    result = asyncio.run(inquiry.format(format_))
+    print(result)  # noqa: T201
+
+
+@cli.command()
+@click.pass_context
+def launch(
+    ctx: click.Context,
+):
+    """
+    Launch MCP server.
+    """
+    registry = ctx.meta["registry"]
+    server_name = ctx.meta["server_name"]
+    if not server_name:
+        raise click.BadParameter("MCP server name is required")
+    servers = registry.select(server_name)
+    servers[0].launch()

cratedb_toolkit/query/mcp/cli.py

@@ -0,0 +1,134 @@
+# Inquire Model Context Protocol (MCP) servers.
+#
+# Usage:
+#
+#   uvx 'cratedb-toolkit[mcp]' query mcp inquire --format=markdown | sponge mcp-cratedb-landscape.md
+#
+# ruff: noqa: T201
+import dataclasses
+import io
+import logging
+import typing as t
+from contextlib import redirect_stdout
+
+from cratedb_toolkit.query.mcp.model import McpServer
+
+from .util import McpServerCapabilities, to_json, to_yaml
+
+logger = logging.getLogger(__name__)
+
+
+@dataclasses.dataclass
+class McpServerInquiry:
+    """
+    Inquire capabilities of MCP server, and format as JSON, Markdown, or YAML.
+    """
+
+    servers: t.List[McpServer]
+
+    title = "Database MCP servers for PostgreSQL and CrateDB"
+    text = (
+        "This page provides an overview about a fragment of the MCP server landscape,\n"
+        "focusing on those that are talking to PostgreSQL and CrateDB databases.\n"
+        "\n"
+        "It enumerates the most popular adapters, and includes their detailed API capabilities.\n"
+    )
+
+    notes = (
+        ":::{note}\n"
+        "This page was generated automatically, please do not edit manually. To rebuild, use this command:\n"
+        "```shell\n"
+        "uvx 'cratedb-toolkit[mcp]' query mcp inquire --format=markdown | sponge doc/query/mcp/landscape.md\n"
+        "```\n"
+        ":::\n"
+    )
+
+    seealso = (
+        ":::{seealso}\n"
+        "Ready-to-run example programs about all the adapters are available at\n"
+        "https://github.com/crate/cratedb-examples/tree/main/framework/mcp.\n"
+        ":::\n"
+    )
+
+    def __post_init__(self):
+        if not self.servers:
+            msg = "No servers selected"
+            logger.error(msg)
+            raise UserWarning(msg)
+
+    @staticmethod
+    async def get_capabilities(server: McpServer):
+        """
+        Launch MCP server with stdio transport, and inquire API for capabilities.
+
+        Derived from:
+        https://github.com/modelcontextprotocol/python-sdk?tab=readme-ov-file#writing-mcp-clients
+        """
+        from mcp import ClientSession, StdioServerParameters
+        from mcp.client.stdio import stdio_client
+
+        if server.program is None:
+            raise ValueError("Program name for MCP server not defined")
+
+        # Create server parameters for stdio connection.
+        server_params = StdioServerParameters(
+            command=server.program,
+            args=server.args,
+            env=server.env,
+        )
+
+        async with stdio_client(server_params) as (read, write):
+            async with ClientSession(read, write) as session:
+                # Initialize the connection.
+                await session.initialize()
+
+                # Inquire API.
+                response = McpServerCapabilities(session)
+                await response.inquire()
+                return response
+
+    async def format(self, variant: str):
+        if variant == "json":
+            return await self.to_json()
+        elif variant == "markdown":
+            return await self.to_markdown()
+        elif variant == "yaml":
+            return await self.to_yaml()
+        else:
+            raise NotImplementedError(f"Output variant not implemented: {variant}")
+
+    async def to_dict(self):
+        payload: t.Dict[str, t.Any] = {
+            "meta": {"title": self.title, "text": self.text, "notes": self.notes, "seealso": self.seealso},
+            "data": {},
+        }
+        for server in self.servers:
+            capabilities = await self.get_capabilities(server)
+            payload["data"][server.name] = {
+                "meta": server.to_dict(),
+                "capabilities": capabilities.to_dict(),
+            }
+        return payload
+
+    async def to_markdown(self):
+        buffer = io.StringIO()
+        with redirect_stdout(buffer):
+            print(f"# {self.title}")
+            print()
+            print(self.text)
+            for server in self.servers:
+                print(server.to_markdown())
+                try:
+                    capabilities = await self.get_capabilities(server)
+                    print(capabilities.to_markdown())
+                except Exception as ex:
+                    logger.error(f"MCP server capability inquiry failed: {ex}")
+            print(self.notes)
+            print(self.seealso)
+        return buffer.getvalue()
+
+    async def to_json(self):
+        return to_json(await self.to_dict())
+
+    async def to_yaml(self):
+        return to_yaml(await self.to_dict())