MCP for a Streaming Database

-- Scenario: a logistics operations agent answering "where is shipment X and will it be late?"
CREATE SOURCE scan_events (
  shipment_id VARCHAR, hub VARCHAR, scan_type VARCHAR,
  next_hub VARCHAR, scanned_at TIMESTAMPTZ,
  WATERMARK FOR scanned_at AS scanned_at - INTERVAL '30 seconds'
) WITH (connector = 'kafka', topic = 'scan_events',
        properties.bootstrap.server = 'broker:9092')
FORMAT PLAIN ENCODE JSON;

CREATE SOURCE ops_pg WITH (
  connector = 'postgres-cdc', hostname = 'pg', port = '5432',
  username = 'rw', password = '...', database.name = 'ops', schema.name = 'public'
);

CREATE TABLE shipments (shipment_id VARCHAR PRIMARY KEY, customer_id BIGINT,
                        origin VARCHAR, destination VARCHAR, promised_by TIMESTAMPTZ,
                        service VARCHAR)
  FROM ops_pg TABLE 'public.shipments';

CREATE TABLE hub_sla (hub VARCHAR PRIMARY KEY, max_dwell INTERVAL)
  FROM ops_pg TABLE 'public.hub_sla';

-- One view per question the agent will be asked.
CREATE MATERIALIZED VIEW shipment_latest AS
SELECT shipment_id, hub, scan_type, next_hub, scanned_at
FROM (SELECT *, ROW_NUMBER() OVER (PARTITION BY shipment_id ORDER BY scanned_at DESC) AS rn
      FROM scan_events) t
WHERE rn = 1;

CREATE MATERIALIZED VIEW shipment_status AS
SELECT s.shipment_id, s.customer_id, s.origin, s.destination, s.service, s.promised_by,
       l.hub AS last_hub, l.scan_type AS last_scan, l.next_hub, l.scanned_at AS last_scan_at,
       (l.scan_type <> 'delivered' AND l.scanned_at + h.max_dwell < s.promised_by) AS on_track
FROM shipments s
LEFT JOIN shipment_latest l ON l.shipment_id = s.shipment_id
LEFT JOIN hub_sla h ON h.hub = l.hub;

CREATE MATERIALIZED VIEW hub_load_15m AS
SELECT hub, window_start, window_end,
       COUNT(*) FILTER (WHERE scan_type = 'arrived')  AS arrivals,
       COUNT(*) FILTER (WHERE scan_type = 'departed') AS departures
FROM HOP(scan_events, scanned_at, INTERVAL '1 minute', INTERVAL '15 minutes')
GROUP BY hub, window_start, window_end;

CREATE INDEX shipment_status_by_id ON shipment_status (shipment_id);
CREATE INDEX shipment_status_by_customer ON shipment_status (customer_id);

-- The catalog is how the agent discovers what to query. Plain language, one row per view.
CREATE TABLE agent_catalog (
  view_name VARCHAR PRIMARY KEY, purpose VARCHAR, lookup_key VARCHAR, freshness VARCHAR
);
INSERT INTO agent_catalog VALUES
  ('shipment_status', 'Current location, last scan and on-track flag for one shipment',
   'shipment_id or customer_id', 'Within a few seconds of the last scan'),
  ('hub_load_15m', 'Arrivals and departures per hub in sliding 15-minute windows',
   'hub; use the row with the latest window_end', 'Window advances every minute'),
  ('shipment_latest', 'Raw latest scan per shipment; prefer shipment_status',
   'shipment_id', 'Within a few seconds');

-- One user per agent. The MCP server has DDL and DML tools; this is what makes them inert.
CREATE USER ops_agent WITH PASSWORD 'change-me';

GRANT SELECT ON MATERIALIZED VIEW shipment_status TO ops_agent;
GRANT SELECT ON MATERIALIZED VIEW hub_load_15m TO ops_agent;
GRANT SELECT ON TABLE agent_catalog TO ops_agent;

-- A second agent, a platform assistant for engineers, gets to read the catalog and explain plans.
CREATE USER platform_assistant WITH PASSWORD 'change-me-too';
GRANT SELECT ON ALL MATERIALIZED VIEWS IN SCHEMA public TO platform_assistant;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO platform_assistant;

-- Verify before handing out the connection string: as ops_agent, a write is refused.
-- psql -h localhost -p 4566 -U ops_agent -d dev -c "DELETE FROM shipments"
-- ERROR: permission denied for table shipments

# Option A: from source, stdio. This is what desktop hosts launch.
git clone https://github.com/risingwavelabs/risingwave-mcp.git
cd risingwave-mcp && pip install -r requirements.txt     # Python 3.10+
export RISINGWAVE_CONNECTION_STR="postgresql://ops_agent:change-me@localhost:4566/dev"
export MCP_TRANSPORT=stdio
python src/main.py

# Option B: Docker, streamable HTTP on :8000/mcp. One server, many clients.
docker run -p 8000:8000 \
  -e RISINGWAVE_CONNECTION_STR="postgresql://ops_agent:change-me@host.docker.internal:4566/dev" \
  -e MCP_TRANSPORT=streamable-http \
  risingwavelabs/risingwave-mcp-server:0.1.0

# Individual parameters work too: RISINGWAVE_HOST, RISINGWAVE_PORT, RISINGWAVE_USER,
# RISINGWAVE_PASSWORD, RISINGWAVE_DATABASE, RISINGWAVE_SSLMODE. MCP_HOST and MCP_PORT set the listener.

// Claude Desktop: claude_desktop_config.json
{
  "mcpServers": {
    "risingwave-ops": {
      "command": "python",
      "args": ["/opt/risingwave-mcp/src/main.py"],
      "env": {
        "RISINGWAVE_CONNECTION_STR": "postgresql://ops_agent:change-me@localhost:4566/dev",
        "MCP_TRANSPORT": "stdio"
      }
    }
  }
}

// Cursor: .cursor/mcp.json (same shape; or point at a shared HTTP server)
{
  "mcpServers": {
    "risingwave-ops": { "url": "http://rw-mcp.internal:8000/mcp" }
  }
}

// VS Code: .vscode/mcp.json
{
  "servers": {
    "risingwave-ops": {
      "type": "stdio",
      "command": "python",
      "args": ["/opt/risingwave-mcp/src/main.py"],
      "env": { "RISINGWAVE_CONNECTION_STR": "postgresql://ops_agent:change-me@localhost:4566/dev" }
    }
  }
}

# pip install "mcp[cli]"
import asyncio
from mcp import Client

ALLOWED = {"run_select_query", "list_materialized_views",
           "describe_materialized_view", "search_docs"}

async def main():
    async with Client("http://localhost:8000/mcp") as mcp:
        tools = (await mcp.list_tools()).tools
        print(len(tools), "tools offered;", len([t for t in tools if t.name in ALLOWED]), "allowed")

        # Discovery: the catalog tells the model which view answers which question.
        cat = await mcp.call_tool("run_select_query",
                                  {"query": "SELECT view_name, purpose, lookup_key FROM agent_catalog"})
        print(cat.content[0].text)

        # A context read: one indexed lookup on a view that is seconds old.
        res = await mcp.call_tool("run_select_query", {
            "query": "SELECT to_jsonb(s) FROM shipment_status s WHERE shipment_id = 'SHP-88213'"})
        if res.is_error:
            raise RuntimeError(res.content[0].text)
        print(res.content[0].text)

        # Anything that is not a SELECT is refused by the server before it reaches RisingWave.
        bad = await mcp.call_tool("run_select_query", {"query": "DELETE FROM shipments"})
        print(bad.content[0].text)   # Error: Only SELECT/WITH queries are allowed ...

asyncio.run(main())

# An agent loop that bridges MCP tools to the Claude API with an allowlist.
import asyncio, json, anthropic
from mcp import Client

ALLOWED = {"run_select_query", "describe_materialized_view"}
SYSTEM = """You are a logistics operations agent. Data lives in RisingWave views that are a few
seconds behind reality. First read agent_catalog to see which view answers the question, then
run one SELECT against it. Never guess a shipment's status; if the view has no row, say so."""

async def run(question: str) -> str:
    llm = anthropic.Anthropic()
    async with Client("http://localhost:8000/mcp") as mcp:
        tools = [{"name": t.name, "description": t.description, "input_schema": t.input_schema}
                 for t in (await mcp.list_tools()).tools if t.name in ALLOWED]
        messages = [{"role": "user", "content": question}]
        while True:
            resp = llm.messages.create(model="claude-sonnet-5-5", max_tokens=1024,
                                       system=SYSTEM, tools=tools, messages=messages)
            messages.append({"role": "assistant", "content": resp.content})
            if resp.stop_reason != "tool_use":
                return "".join(b.text for b in resp.content if b.type == "text")
            results = []
            for block in resp.content:
                if block.type != "tool_use":
                    continue
                out = await mcp.call_tool(block.name, block.input)
                results.append({"type": "tool_result", "tool_use_id": block.id,
                                "content": out.content[0].text, "is_error": out.is_error})
            messages.append({"role": "user", "content": results})

print(asyncio.run(run("Where is SHP-88213 and will it make its promised delivery?")))

-- What is the agent running right now? Sessions and their current statements.
SHOW PROCESSLIST;

-- Lookups the agent repeats deserve an index; check the plan it gets.
EXPLAIN SELECT to_jsonb(s) FROM shipment_status s WHERE customer_id = 4711;

-- Is the view keeping up? Streaming jobs and their progress.
SHOW JOBS;

-- State size per view, to catch a context view that grew without bound.
SELECT * FROM rw_catalog.rw_table_stats ORDER BY total_key_count DESC;