MCP: Add example about PG-MCP

Author: amotl

framework/mcp/.gitignore

@@ -1 +1,2 @@
 uv.lock
+pg-mcp

framework/mcp/backlog.md

@@ -9,6 +9,8 @@
   https://github.com/crate/crate/issues/17393
 - DBHub: Reading resource `tables` does not work,
   because `WHERE table_schema = 'public'`
+- PG-MCP: Improve installation after packaging has been improved.
+  https://github.com/stuzero/pg-mcp/issues/10
 
 ## Iteration +2
 - General: Evaluate all connectors per `stdio` and `sse`, where possible

framework/mcp/example_pg_mcp.py

@@ -0,0 +1,70 @@
+# PG-MCP Model Context Protocol Server for CrateDB
+# https://github.com/stuzero/pg-mcp
+# https://github.com/crate-workbench/pg-mcp
+#
+# Derived from:
+# https://github.com/modelcontextprotocol/python-sdk?tab=readme-ov-file#writing-mcp-clients
+from cratedb_toolkit.util import DatabaseAdapter
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+import where
+
+from mcp_utils import McpDatabaseConversation
+
+
+async def run():
+    # Create server parameters for stdio connection.
+    server_params = StdioServerParameters(
+        command=where.first("python"),
+        args=["example_pg_mcp_server.py"],
+        env={},
+    )
+
+    async with stdio_client(server_params) as (read, write):
+        async with ClientSession(
+            read, write
+        ) as session:
+            # Initialize the connection.
+            await session.initialize()
+
+            client = McpDatabaseConversation(session)
+            await client.inquire()
+
+            print("## MCP server conversations")
+            print()
+
+            # Provision database content.
+            db = DatabaseAdapter("crate://crate@localhost:4200/")
+            db.run_sql("CREATE TABLE IF NOT EXISTS mcp_pg_mcp (id INT, data TEXT)")
+            db.run_sql("INSERT INTO mcp_pg_mcp (id, data) VALUES (42, 'Hotzenplotz')")
+            db.refresh_table("mcp_pg_mcp")
+
+            # Call a few tools.
+            connection_string = "postgresql://crate@localhost/doc"
+
+            # Connect to the database, receiving the connection UUID.
+            response = await client.call_tool("connect", arguments={"connection_string": connection_string})
+            conn_id = client.decode_json_text(response)["conn_id"]
+
+            # Query and explain, using the connection id.
+            await client.call_tool("pg_query", arguments={
+                "query": "SELECT * FROM sys.summits ORDER BY height DESC LIMIT 3",
+                "conn_id": conn_id,
+            })
+            await client.call_tool("pg_explain", arguments={
+                "query": "SELECT * FROM mcp_pg_mcp",
+                "conn_id": conn_id,
+            })
+
+            # Read a few resources.
+            await client.read_resource(f"pgmcp://{conn_id}/schemas")
+            await client.read_resource(f"pgmcp://{conn_id}/schemas/sys/tables")
+
+            # Disconnect again.
+            await client.call_tool("disconnect", arguments={"conn_id": conn_id,})
+
+
+if __name__ == "__main__":
+    import asyncio
+
+    asyncio.run(run())

framework/mcp/example_pg_mcp_install.sh

@@ -0,0 +1,13 @@
+#!/bin/sh
+# Acquire MCP server part of `pg-mcp`.
+# https://github.com/stuzero/pg-mcp
+
+# FIXME: Improve installation after packaging has been improved.
+#        https://github.com/stuzero/pg-mcp/issues/10
+
+rm -rf pg-mcp
+git clone --depth 1 --no-checkout --filter=blob:none \
+  https://github.com/crate-workbench/pg-mcp.git
+cd pg-mcp
+git checkout 16d7f61d5b3197777293ebae33b519f14a9d6e55 -- pyproject.toml uv.lock server test.py
+uv pip install .

framework/mcp/example_pg_mcp_server.py

@@ -0,0 +1,8 @@
+if __name__ == "__main__":
+    # FIXME: Improve invocation after packaging has been improved.
+    #        https://github.com/stuzero/pg-mcp/issues/10
+    from server.app import logger, mcp
+
+    # TODO: Bring flexible invocation (sse vs. stdio) to mainline.
+    logger.info("Starting MCP server with STDIO transport")
+    mcp.run(transport="stdio")

framework/mcp/mcp_utils.py

@@ -1,5 +1,7 @@
 import io
 import json
+import logging
+
 import mcp.types as types
 from typing import Any
 
@@ -9,13 +11,20 @@
 from pydantic import AnyUrl
 
 
+logger = logging.getLogger(__name__)
+
+
 class McpDatabaseConversation:
     """
     Wrap database conversations through MCP servers.
     """
     def __init__(self, session: ClientSession):
         self.session = session
 
+    @staticmethod
+    def decode_json_text(thing):
+        return json.loads(thing.content[0].text)
+
     def decode_items(self, items):
         return list(map(self.decode_item, json.loads(pydantic_core.to_json(items))))
 
@@ -36,26 +45,38 @@ def list_items(self, items):
             buffer.write("```\n")
         return buffer.getvalue()
 
+    async def entity_info(self, fun, attribute):
+        try:
+            return self.list_items(getattr(await fun(), attribute))
+        except McpError as e:
+            logger.error(f"Not implemented on this server: {e}")
+
+    @staticmethod
+    def dump_info(results):
+        if results:
+            print(results)
+        print()
+
     async def inquire(self):
         print("# MCP server inquiry")
         print()
 
         # List available prompts
         print("## Prompts")
-        try:
-            print(self.list_items((await self.session.list_prompts()).prompts))
-        except McpError as e:
-            print(f"Not implemented on this server: {e}")
-        print()
+        self.dump_info(self.entity_info(self.session.list_prompts, "prompts"))
 
-        # List available resources
+        # List available resources and resource templates
         print("## Resources")
-        print(self.list_items((await self.session.list_resources()).resources))
+        self.dump_info(self.entity_info(self.session.list_resources, "resources"))
+        print()
+
+        print("## Resource templates")
+        self.dump_info(self.entity_info(self.session.list_resource_templates, "resourceTemplates"))
         print()
 
         # List available tools
         print("## Tools")
-        print(self.list_items((await self.session.list_tools()).tools))
+        self.dump_info(self.entity_info(self.session.list_tools, "tools"))
         print()
 
     async def call_tool(

framework/mcp/test.py

@@ -150,3 +150,39 @@ def test_mcp_alchemy():
     assert b"Calling tool: schema_definitions" in p.stdout
     assert b"id: INTEGER, nullable" in p.stdout
     assert b"data: VARCHAR, nullable" in p.stdout
+
+
+@pytest.mark.skipif(sys.version_info < (3, 13), reason="requires Python 3.13+")
+def test_pg_mcp():
+    """
+    Validate the PG-MCP server works well.
+
+    It is written in Python and uses pgasync.
+    https://github.com/crate-workbench/pg-mcp
+    """
+    p = run(f"sh example_pg_mcp_install.sh")
+    print(p.stdout)
+    print(p.stderr)
+    assert p.returncode == 0
+    p = run(f"{sys.executable} example_pg_mcp.py")
+    assert p.returncode == 0
+
+    # Validate output specific to the MCP server.
+    assert b"Processing request of type" in p.stderr
+    assert b"PromptListRequest" in p.stderr
+    assert b"ListResourcesRequest" in p.stderr
+    assert b"ListToolsRequest" in p.stderr
+    assert b"CallToolRequest" in p.stderr
+
+    # Validate output specific to CrateDB.
+    assert b"Calling tool: pg_query" in p.stdout
+    assert b"mountain: Mont Blanc" in p.stdout
+
+    assert b"Calling tool: pg_explain" in p.stdout
+
+    assert b"Reading resource: pgmcp://" in p.stdout
+    assert b"schema_name: blob" in p.stdout
+    assert b"schema_name: doc" in p.stdout
+    assert b"schema_name: sys" in p.stdout
+    assert b"table_name: jobs" in p.stdout
+    assert b"table_name: shards" in p.stdout