MCP: Add example about PG-MCP

Author: amotl

.github/workflows/framework-mcp.yml

@@ -83,6 +83,11 @@ jobs:
 
       - name: Install utilities
         run: |
+
+          # Install sponge.
+          sudo apt-get install moreutils
+
+          # Install pueblo.
           uv pip install -r requirements.txt
 
       - name: Validate framework/mcp

framework/mcp/.gitignore

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

framework/mcp/README.md

@@ -40,6 +40,12 @@ program.
   server package, providing a range of tools. It is written in Python, and uses
   [SQLAlchemy] and the [CrateDB SQLAlchemy dialect].
 
+- `example_pg_mcp.py`:
+  The [PG-MCP] server is specialised to talk to PostgreSQL servers. With a few adjustments,
+  the adapter can also talk to CrateDB. The project offers rich MCP server capabilities,
+  and includes advanced client programs for Claude and Gemini that work out of the box.
+  It is written in Python, optionally to be invoked with `uv` or `uvx`.
+
 ## Resources
 
 - Read a [brief introduction to MCP] by ByteByteGo.
@@ -142,6 +148,7 @@ unlocking more details and features.
 [@modelcontextprotocol/server-postgres]: https://www.npmjs.com/package/@modelcontextprotocol/server-postgres
 [npx]: https://docs.npmjs.com/cli/v11/commands/npx
 [oterm configuration]: https://ggozad.github.io/oterm/tools/mcp/
+[PG-MCP]: https://github.com/stuzero/pg-mcp-server
 [quarkus-mcp-servers]: https://github.com/quarkiverse/quarkus-mcp-servers
 [SQLAlchemy]: https://sqlalchemy.org/
 [uv]: https://docs.astral.sh/uv/

framework/mcp/backlog.md

@@ -9,6 +9,13 @@
   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-server/issues/10
+- PG-MCP: Resource `pgmcp://{conn_id}/` makes `asyncpg` invoke
+  `WITH RECURSIVE typeinfo_tree`, which fails on CrateDB.
+  - https://github.com/crate/crate/issues/11757
+  - https://github.com/crate/crate/issues/12544
+- PG-MCP: Fix `/rowcount endpoint`
 
 ## Iteration +2
 - General: Evaluate all connectors per `stdio` and `sse`, where possible

framework/mcp/example_pg_mcp.py

@@ -0,0 +1,87 @@
+# PG-MCP Model Context Protocol Server for CrateDB
+# https://github.com/stuzero/pg-mcp-server
+# https://github.com/crate-workbench/pg-mcp-server
+#
+# 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.
+            schema = "sys"
+            table = "summits"
+            await client.read_resource(f"pgmcp://{conn_id}/")
+            await client.read_resource(f"pgmcp://{conn_id}/schemas")
+            await client.read_resource(f"pgmcp://{conn_id}/schemas/{schema}/tables")
+            await client.read_resource(f"pgmcp://{conn_id}/schemas/{schema}/tables/{table}/columns")
+            #await client.read_resource(f"pgmcp://{conn_id}/schemas/{schema}/tables/{table}/indexes")
+            #await client.read_resource(f"pgmcp://{conn_id}/schemas/{schema}/tables/{table}/constraints")
+            #await client.read_resource(f"pgmcp://{conn_id}/schemas/{schema}/tables/{table}/indexes/{index}")
+            #await client.read_resource(f"pgmcp://{conn_id}/schemas/{schema}/tables/{table}/constraints/{constraint}")
+            #await client.read_resource(f"pgmcp://{conn_id}/schemas/{schema}/extensions")
+            #await client.read_resource(f"pgmcp://{conn_id}/schemas/{schema}/extensions/{extension}")
+            await client.read_resource(f"pgmcp://{conn_id}/schemas/{schema}/tables/{table}/sample")
+            await client.read_resource(f"pgmcp://{conn_id}/schemas/{schema}/tables/{table}/rowcount")
+
+            # Invoke a prompt.
+            await client.get_prompt("nl_to_sql_prompt", arguments={
+                "query": "Give me 5 Austria mountains",
+            })
+
+            # 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,21 @@
+#!/bin/sh
+# Acquire MCP server part of `pg-mcp`.
+# https://github.com/stuzero/pg-mcp-server
+# https://github.com/crate-workbench/pg-mcp-server
+
+# FIXME: Improve installation after packaging has been improved.
+#        https://github.com/stuzero/pg-mcp-server/issues/10
+
+set -e
+TARGET="/tmp/pg-mcp-server"
+rm -rf ${TARGET}
+git clone --depth 1 --no-checkout --filter=blob:none \
+  https://github.com/crate-workbench/pg-mcp-server.git \
+  ${TARGET}
+cd ${TARGET}
+git checkout 82733d1a886bf1a14592d1fbb305205901f2bb35 -- pyproject.toml uv.lock server test.py
+cat pyproject.toml | grep -v requires-python | sponge pyproject.toml
+uv pip install .
+
+# /Users/amo/dev/crate-workbench/sources/pg-mcp-server
+# https://github.com/crate-workbench/pg-mcp.git

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-server/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(await 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(await self.entity_info(self.session.list_resources, "resources"))
+        print()
+
+        print("## Resource templates")
+        self.dump_info(await 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(await self.entity_info(self.session.list_tools, "tools"))
         print()
 
     async def call_tool(

framework/mcp/test.py

@@ -150,3 +150,43 @@ 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
+    """
+
+    # FIXME: Manually invoke pre-installation step.
+    p = run(f"sh example_pg_mcp_install.sh")
+    assert p.returncode == 0, p.stderr
+
+    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
+
+    assert b"Getting prompt: nl_to_sql_prompt" in p.stdout
+    assert b"You are an expert PostgreSQL database query assistant" in p.stdout