Implementation:MaterializeInc Materialize MCP Server

Overview

The MCP Materialize Server is a Model Context Protocol (MCP) server that dynamically exposes Materialize indexes as tools. Each Materialize index that the connected role is authorized to SELECT from (and whose cluster allows USAGE) is surfaced as a tool whose inputs correspond to the indexed columns and whose output is the remaining columns of the underlying view.

The server is implemented across several modules in misc/mcp-materialize/mcp_materialize/.

Architecture

Module	Purpose
__init__.py	Server entry point, connection pool lifecycle, MCP handler registration
config.py	Configuration dataclass and CLI argument parsing
mz_client.py	Materialize client wrapping tool discovery and execution logic
transports.py	Transport layer implementations (stdio, SSE, HTTP)
mzcompose.py	Compose-based integration test workflow

Server Entry Point

The __init__.py module serves as the main entry point. It initializes an asynchronous connection pool using psycopg_pool.AsyncConnectionPool, creates an MzClient instance, and registers MCP protocol handlers.

Connection Pool Lifecycle

The create_client async context manager performs the following:

1. Creates an AsyncConnectionPool with configurable min_size and max_size.
2. Configures each connection with autocommit=True and the application name mcp_materialize.
3. Validates connectivity by executing a test query that retrieves the environment ID and current role.
4. Yields an MzClient instance wrapping the pool.

@asynccontextmanager
async def create_client(cfg: Config) -> AsyncIterator[MzClient]:
    async with AsyncConnectionPool(
        conninfo=cfg.dsn,
        min_size=cfg.pool_min_size,
        max_size=cfg.pool_max_size,
        kwargs={"application_name": "mcp_materialize"},
        configure=configure,
    ) as pool:
        async with MzClient(pool=pool) as client:
            yield client

MCP Handler Registration

The run function registers two core MCP handlers on the mcp.server.Server instance:

• list_tools -- Delegates to MzClient.list_tools() to dynamically enumerate available Materialize indexes as MCP tools.
• call_tool -- Delegates to MzClient.call_tool(name, arguments) to execute a parameterized SELECT against the indexed view. On error, it notifies the client that the tool list has changed.

The server enables tools_changed notifications so that clients are informed when the set of available tools changes dynamically.

Configuration

The config.py module defines a frozen Config dataclass and a load_config() function that parses CLI arguments with environment variable fallbacks.

Parameter	CLI Flag	Env Var	Default
dsn	--mz-dsn	MZ_DSN	postgresql://materialize@localhost:6875/materialize
transport	--transport	MCP_TRANSPORT	stdio
host	--host	MCP_HOST	0.0.0.0
port	--port	MCP_PORT	3001 (SSE) / 8001 (HTTP)
pool_min_size	--pool-min-size	MCP_POOL_MIN_SIZE	1
pool_max_size	--pool-max-size	MCP_POOL_MAX_SIZE	10
log_level	--log-level	MCP_LOG_LEVEL	INFO

The transport choices are stdio, http, and sse. The default port depends on which transport is selected.

MzClient

The mz_client.py module implements the MzClient class which manages the mapping between Materialize indexes and MCP tools.

MzTool

Each discovered index is represented as an MzTool instance with the following attributes:

Attribute	Description
name	Tool identifier derived from the index
database	Database containing the indexed view
schema	Schema containing the indexed view
object_name	Name of the underlying view
cluster	Cluster where the index resides
title	Human-readable title for the tool
description	Auto-generated description of the tool
input_schema	JSON Schema for indexed columns (inputs)
output_schema	JSON Schema for remaining columns (outputs)
output_columns	List of output column names

The as_tool() method converts an MzTool into an MCP Tool object with readOnlyHint=True annotations.

Tool Discovery and Subscription

On initialization (__aenter__), MzClient:

1. Loads the tool catalog by executing a SQL query from tools.sql.
2. Starts a background asyncio.Task that subscribes to catalog changes, ensuring the tool list stays current.

The tool dictionary is protected by an aiorwlock.RWLock to allow concurrent reads during tool execution while serializing writes during catalog refreshes.

MissingTool Exception

A MissingTool exception is raised when call_tool is invoked with a tool name that does not exist in the current catalog.

Transport Layer

The transports.py module provides three async transport implementations:

stdio Transport

Uses mcp.stdio_server to exchange JSON messages over stdin/stdout. This is the default transport, suitable for local CLI integration.

async def stdio_transport(server: Server, options: InitializationOptions):
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, options)

SSE Transport

Implements server-sent events over HTTP using Starlette and Uvicorn. Exposes two routes:

Route	Method	Purpose
/sse	GET	Establishes the SSE connection
/messages/	POST	Receives client messages

HTTP Transport

Uses StreamableHTTPSessionManager in stateless mode. Mounts the MCP handler at /mcp and runs via Uvicorn with a Starlette application lifecycle.

session_manager = StreamableHTTPSessionManager(
    app=server,
    stateless=True,
)

Integration Testing

The mzcompose.py file defines a compose-based test workflow for the MCP server.

Services

Service	Description
Materialized	Materialize instance with configurable options
Mcp	MCP server container used to run pytest

Test Workflow

The workflow_default function:

1. Accepts optional filter (filter), keyword (-k), and output suppression (-s) arguments.
2. Iterates through test cases (currently a single mcp-server case).
3. Configures a Materialized service with default_replication_factor=1.
4. Brings up the Materialized service and runs uv run pytest tests/ inside the MCP container, passing the DSN via the MZ_DSN environment variable.

c.run(
    "mcp",
    "uv", "run", "pytest", *test_args,
    env_extra={
        "MZ_DSN": "postgres://materialize@materialized:6875/materialize",
    },
)

Key Source Files

File	Path
Server entry point	misc/mcp-materialize/mcp_materialize/__init__.py
Configuration	misc/mcp-materialize/mcp_materialize/config.py
Materialize client	misc/mcp-materialize/mcp_materialize/mz_client.py
Transport implementations	misc/mcp-materialize/mcp_materialize/transports.py
Integration test compose	misc/mcp-materialize/mzcompose.py