You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "viirya (via GitHub)" <gi...@apache.org> on 2026/05/01 17:04:32 UTC
[PR] [SPARK-XXXXX][PYTHON] Add Spark MCP (Model Context Protocol) server [spark]
viirya opened a new pull request, #55648:
URL: https://github.com/apache/spark/pull/55648
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
Adds an Apache Spark MCP server that acts as a thin client over Spark Connect, exposing Spark capabilities as MCP tools for LLM consumption. Catalog browsing, SQL execution, and query plan tools are read-only by default.
Module layout (python/pyspark/sql/mcp/):
- server.py CLI entry point and MCP tool registration
- config.py ServerConfig dataclass (env + CLI sources)
- session.py Lazy SparkSession holder over Spark Connect
- safety.py Read-only SQL guardrail
- tools/registry.py Tool spec / handler abstraction
- tools/session.py get_session_info (with config redaction)
- tools/catalog.py list_catalogs, list_databases, list_tables, describe_table
- tools/query.py list_functions, execute_sql, preview_table, explain_query, analyze_query
Tool handlers are MCP-SDK-agnostic and Connect-import-free at module load time, so the unit tests run without grpcio or the mcp SDK installed. 14 unit tests in python/pyspark/sql/tests/mcp/test_mcp_tools.py exercise the full tool surface against an in-memory fake session.
### Why are the changes needed?
LLM clients can already talk to MCP servers; Spark Connect already separates client from cluster. This module connects the two: a Spark cluster shows up to an LLM as a set of safe, paginated tools — `list_tables`, `describe_table`, `execute_sql`, `explain_query`, etc.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
Yes. User can do Spark queries in LLMs like Claude Code using these MCP tools.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Unit test. Manual test in Claude Code.
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
Generated-by: Claude Code
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56698][PYTHON] Add Spark MCP (Model Context Protocol) server [spark]
Posted by "holdenk (via GitHub) via reviews" <re...@spark.apache.org>.
holdenk commented on PR #55648:
URL: https://github.com/apache/spark/pull/55648#issuecomment-6536958372
Any progress on the SPIP @viirya ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56698][PYTHON] Add Spark MCP (Model Context Protocol) server [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on PR #55648:
URL: https://github.com/apache/spark/pull/55648#issuecomment-4360832289
Personally I think this requires an SPIP. This is a completely new feature that requires some maintenance effort.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56698][PYTHON] Add Spark MCP (Model Context Protocol) server [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #55648: [SPARK-56698][PYTHON] Add Spark MCP (Model Context Protocol) server
URL: https://github.com/apache/spark/pull/55648
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56698][PYTHON] Add Spark MCP (Model Context Protocol) server [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #55648:
URL: https://github.com/apache/spark/pull/55648#issuecomment-5389321120
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56698][PYTHON] Add Spark MCP (Model Context Protocol) server [spark]
Posted by "spark-test (via GitHub) via reviews" <re...@spark.apache.org>.
spark-test commented on code in PR #55648:
URL: https://github.com/apache/spark/pull/55648#discussion_r4509112098
##########
python/pyspark/sql/mcp/tools/catalog.py:
##########
@@ -0,0 +1,218 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Catalog browsing tools."""
+
+from __future__ import annotations
+
+from dataclasses import asdict, is_dataclass
+from typing import Any, Dict, List
+
+from pyspark.sql.mcp.session import SessionHolder
+from pyspark.sql.mcp.tools.registry import ToolSpec
+
+
+# ---------------------------------------------------------------------------
+# Result shaping
+# ---------------------------------------------------------------------------
+
+
+def _to_dict(obj: Any) -> Dict[str, Any]:
+ """Best-effort conversion of catalog dataclasses (CatalogMetadata, Database,
+ Table, Column, Function) into plain dicts."""
+ if is_dataclass(obj) and not isinstance(obj, type):
+ return asdict(obj)
+ if hasattr(obj, "_asdict"): # NamedTuple
+ return dict(obj._asdict())
+ if isinstance(obj, dict):
+ return obj
+ # Fallback: scrape public attributes.
+ return {
+ k: getattr(obj, k)
+ for k in dir(obj)
+ if not k.startswith("_") and not callable(getattr(obj, k))
+ }
+
+
+def _paginate(items: List[Any], limit: int, offset: int) -> Dict[str, Any]:
+ total = len(items)
+ end = offset + limit if limit is not None else total
+ page = items[offset:end]
+ return {
+ "items": [_to_dict(x) for x in page],
+ "offset": offset,
+ "limit": limit,
+ "returned": len(page),
+ "total": total,
+ "truncated": end < total,
+ }
+
+
+# ---------------------------------------------------------------------------
+# list_catalogs
+# ---------------------------------------------------------------------------
+
+
+async def _handle_list_catalogs(args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ spark = holder.get()
+ catalogs = list(spark.catalog.listCatalogs())
+ limit = int(args.get("limit", holder.config.default_page_size))
+ offset = int(args.get("offset", 0))
+ return _paginate(catalogs, limit, offset)
+
+
+def list_catalogs_spec() -> ToolSpec:
+ return ToolSpec(
+ name="list_catalogs",
+ description="List catalogs available to the current Spark session.",
+ input_schema={
+ "type": "object",
+ "properties": {
+ "limit": {"type": "integer", "minimum": 1, "maximum": 1000, "default": 100},
+ "offset": {"type": "integer", "minimum": 0, "default": 0},
+ },
+ "additionalProperties": False,
+ },
+ handler=_handle_list_catalogs,
+ )
+
+
+# ---------------------------------------------------------------------------
+# list_databases
+# ---------------------------------------------------------------------------
+
+
+async def _handle_list_databases(args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ spark = holder.get()
+ catalog = args.get("catalog")
+ pattern = args.get("pattern")
+
+ if catalog:
+ spark.catalog.setCurrentCatalog(catalog)
+ if pattern is not None:
+ databases = list(spark.catalog.listDatabases(pattern))
+ else:
+ databases = list(spark.catalog.listDatabases())
+
+ limit = int(args.get("limit", holder.config.default_page_size))
+ offset = int(args.get("offset", 0))
+ return _paginate(databases, limit, offset)
+
+
+def list_databases_spec() -> ToolSpec:
+ return ToolSpec(
+ name="list_databases",
+ description=(
+ "List databases (a.k.a. schemas) in the current or named catalog. "
+ "Optional name pattern uses Spark's LIKE semantics."
+ ),
+ input_schema={
+ "type": "object",
+ "properties": {
+ "catalog": {"type": "string"},
+ "pattern": {"type": "string"},
+ "limit": {"type": "integer", "minimum": 1, "maximum": 1000, "default": 100},
+ "offset": {"type": "integer", "minimum": 0, "default": 0},
+ },
+ "additionalProperties": False,
+ },
+ handler=_handle_list_databases,
+ )
+
+
+# ---------------------------------------------------------------------------
+# list_tables
+# ---------------------------------------------------------------------------
+
+
+async def _handle_list_tables(args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ spark = holder.get()
+ database = args.get("database")
+ pattern = args.get("pattern")
+
+ kwargs: Dict[str, Any] = {}
+ if database is not None:
+ kwargs["dbName"] = database
+ if pattern is not None:
+ kwargs["pattern"] = pattern
+ tables = list(spark.catalog.listTables(**kwargs))
+
+ limit = int(args.get("limit", holder.config.default_page_size))
+ offset = int(args.get("offset", 0))
+ return _paginate(tables, limit, offset)
+
+
+def list_tables_spec() -> ToolSpec:
+ return ToolSpec(
+ name="list_tables",
+ description=(
+ "List tables and views in a database. Defaults to the current "
+ "database if 'database' is omitted. Returns each table's name, "
+ "type (TABLE/VIEW/TEMPORARY), description, and database."
Review Comment:
**Nit (P3):** `Table._asdict()` yields `name`, `catalog`, `namespace`, `description`, `tableType` and `isTemporary`. `database` is a `@property` and never appears, and Spark's `tableType` values are MANAGED, EXTERNAL, VIEW and TEMPORARY, never TABLE. describe_table's description (line 202) promises `database` too. An LLM that trusts these descriptions looks for fields that are not there; listing the real fields fixes it.
**Verification:**
- **Inspection:** Every field and value named in the two descriptions appears in the items the tools return.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56698][PYTHON] Add Spark MCP (Model Context Protocol) server [spark]
Posted by "franciscoabsampaio (via GitHub)" <gi...@apache.org>.
franciscoabsampaio commented on PR #55648:
URL: https://github.com/apache/spark/pull/55648#issuecomment-4461404948
"—" 🧐
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56698][PYTHON] Add Spark MCP (Model Context Protocol) server [spark]
Posted by "holdenk (via GitHub) via reviews" <re...@spark.apache.org>.
holdenk commented on code in PR #55648:
URL: https://github.com/apache/spark/pull/55648#discussion_r4429745651
##########
python/pyspark/sql/mcp/safety.py:
##########
@@ -0,0 +1,53 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import re
+
+from pyspark.errors import PySparkValueError
+
+# Read-only allow-list. Comments and leading whitespace are stripped before
+# matching. We deliberately keep this conservative — anything not on the list
+# is rejected when read_only is on.
+_READ_ONLY_LEAD = re.compile(
+ r"^(select|with|show|describe|desc|explain|use|set|reset)\b",
+ re.IGNORECASE,
+)
+
+# Strip /* ... */ block comments and -- line comments before matching.
+_BLOCK_COMMENT = re.compile(r"/\*.*?\*/", re.DOTALL)
+_LINE_COMMENT = re.compile(r"--[^\n]*")
+
+
+def assert_read_only(query: str) -> None:
+ """Reject queries that mutate state when the server is in read-only mode.
+
+ The check is best-effort and string-based; it is a guardrail, not a
+ security boundary. Real isolation belongs at the Spark cluster (catalog
+ permissions, workspace policies, etc.).
+ """
+ stripped = _LINE_COMMENT.sub("", _BLOCK_COMMENT.sub("", query)).strip()
+ if not stripped:
+ raise PySparkValueError(
+ errorClass="MCP_EMPTY_QUERY",
+ messageParameters={},
+ )
+ if not _READ_ONLY_LEAD.match(stripped):
+ first_token = stripped.split(None, 1)[0]
+ raise PySparkValueError(
+ errorClass="MCP_READ_ONLY_VIOLATION",
+ messageParameters={"statement": first_token},
+ )
Review Comment:
That this is best effort protection doesn't really come through (also maybe better/easier to enforce in the Scala code)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56698][PYTHON] Add Spark MCP (Model Context Protocol) server [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #55648:
URL: https://github.com/apache/spark/pull/55648#issuecomment-4377110887
Thanks @gaogaotiantian and @franciscoabsampaio for the feedback — agreed on both counts.
I should have started with an SPIP. This is user-facing surface area that introduces a new module, an optional third-party SDK dependency (`mcp`), and a permanent set of tool-shape contracts that the project will own going forward. Those are exactly the criteria the SPIP process is meant to gate.
Plan from here:
1. I'll file the SPIP on dev@ and JIRA, with this PR linked as a reference implementation. A working draft is already written; I'll share it on the SPIP thread so people can react to the design before deep code review starts here.
2. The PR itself is functionally ready -- I'll keep it open at its current status so reviewers who want to look at the code alongside the SPIP discussion can do so, but I'm not asking for review merge-readiness until the SPIP lands.
3. Once the SPIP is accepted (or revised), I'll address review comments here (including documentation — there is a module README at `python/pyspark/sql/mcp/README.md` today, but no user-facing Sphinx page yet, which is on me to add), and rebase.
@franciscoabsampaio — quick note on `asyncio.to_thread`: it's there because the MCP Python SDK's server loop is asyncio-native, and the Spark Connect Python client API is synchronous (`df.collect()`, `df._explain_string(...)`, etc.). To honour a `--query-timeout-seconds` budget without blocking the asyncio event loop, the tool handlers offload the blocking Connect calls via `asyncio.to_thread` and wrap them in `asyncio.wait_for`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56698][PYTHON] Add Spark MCP (Model Context Protocol) server [spark]
Posted by "franciscoabsampaio (via GitHub)" <gi...@apache.org>.
franciscoabsampaio commented on PR #55648:
URL: https://github.com/apache/spark/pull/55648#issuecomment-4372653618
Hi
First time reviewing a PR. Before any ado, thanks for the contribution.
Firstly, I'm concerned with the lack of documentation - this is a rather large feature.
Secondly, upon very brief inspection, I see some patterns that I don't find anywhere else in the spark codebase, namely, the usage of `asyncio.to_thread()`.
Like gaogaotiantian mentioned, this most likely needs a SPIP first.
Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56698][PYTHON] Add Spark MCP (Model Context Protocol) server [spark]
Posted by "spark-test (via GitHub) via reviews" <re...@spark.apache.org>.
spark-test commented on code in PR #55648:
URL: https://github.com/apache/spark/pull/55648#discussion_r4509112039
##########
python/pyspark/sql/mcp/tools/query.py:
##########
@@ -0,0 +1,363 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""SQL execution and plan tools."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Awaitable, Callable, Dict, List, Optional, TypeVar
+
+from pyspark.errors import PySparkRuntimeError, PySparkValueError
+from pyspark.sql.mcp.safety import assert_read_only
+from pyspark.sql.mcp.session import SessionHolder
+from pyspark.sql.mcp.tools.catalog import _to_dict
+from pyspark.sql.mcp.tools.registry import ToolSpec
+
+
+T = TypeVar("T")
+
+
+async def _with_timeout(holder: SessionHolder, func: Callable[[], T], *, label: str) -> T:
+ """Run a blocking Spark Connect call in a worker thread under the configured timeout."""
+ timeout = holder.config.query_timeout_seconds
+ coro: Awaitable[T] = asyncio.to_thread(func)
+ if timeout and timeout > 0:
+ try:
+ return await asyncio.wait_for(coro, timeout=timeout)
+ except asyncio.TimeoutError as exc:
+ raise PySparkRuntimeError(
+ errorClass="MCP_QUERY_TIMEOUT",
+ messageParameters={"label": label, "timeout": str(timeout)},
+ ) from exc
+ return await coro
+
+
+# ---------------------------------------------------------------------------
+# list_functions
+# ---------------------------------------------------------------------------
+
+
+async def _handle_list_functions(args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ spark = holder.get()
+ database = args.get("database")
+ pattern = args.get("pattern")
+
+ kwargs: Dict[str, Any] = {}
+ if database is not None:
+ kwargs["dbName"] = database
+ if pattern is not None:
+ kwargs["pattern"] = pattern
+ funcs = list(spark.catalog.listFunctions(**kwargs))
+
+ limit = int(args.get("limit", holder.config.default_page_size))
+ offset = int(args.get("offset", 0))
+ total = len(funcs)
+ page = funcs[offset : offset + limit]
+ return {
+ "items": [_to_dict(f) for f in page],
+ "offset": offset,
+ "limit": limit,
+ "returned": len(page),
+ "total": total,
+ "truncated": offset + limit < total,
+ }
+
+
+def list_functions_spec() -> ToolSpec:
+ return ToolSpec(
+ name="list_functions",
+ description=(
+ "List user-defined and built-in functions visible in a database. "
+ "Useful when authoring SQL — surfaces both UDFs and the built-in "
+ "function library."
+ ),
+ input_schema={
+ "type": "object",
+ "properties": {
+ "database": {"type": "string"},
+ "pattern": {"type": "string"},
+ "limit": {"type": "integer", "minimum": 1, "maximum": 1000, "default": 100},
+ "offset": {"type": "integer", "minimum": 0, "default": 0},
+ },
+ "additionalProperties": False,
+ },
+ handler=_handle_list_functions,
+ )
+
+
+# ---------------------------------------------------------------------------
+# Result rendering helpers
+# ---------------------------------------------------------------------------
+
+
+def _row_to_dict(row: Any) -> Dict[str, Any]:
+ if hasattr(row, "asDict"):
+ return row.asDict(recursive=True)
Review Comment:
**Blocking (P1):** `row.asDict(recursive=True)` is `dict(zip(__fields__, values))`, so for a repeated column name it keeps only the last value, even though Connect delivers every value positionally. For `SELECT o.id, c.id ...`, or `SELECT *` over a join, the JSON rows silently drop a column while `schema` still lists both. In markdown, `_render_markdown`'s `row.get(name)` shows the second value under both headers. Serializing by position against the schema, with disambiguated keys, keeps every value.
**Verification:**
- **Regression:** A query returning two columns named id with different values returns both values, in JSON rows and in the markdown table.
- **Behavior:** Queries whose column names are unique produce the same JSON and markdown output as before.
##########
python/pyspark/sql/mcp/safety.py:
##########
@@ -0,0 +1,53 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import re
+
+from pyspark.errors import PySparkValueError
+
+# Read-only allow-list. Comments and leading whitespace are stripped before
+# matching. We deliberately keep this conservative — anything not on the list
+# is rejected when read_only is on.
+_READ_ONLY_LEAD = re.compile(
+ r"^(select|with|show|describe|desc|explain|use|set|reset)\b",
Review Comment:
**Blocking (P1):** `with` in this allow-list admits more than queries. Spark's grammar accepts a CTE prefix on DML (`ctes? dmlStatementNoWith`, SqlBaseParser.g4:227), so `WITH s AS (SELECT 1 AS id) INSERT INTO t SELECT * FROM s` passes this check; running these regexes on the DELETE, UPDATE and MERGE forms accepts them too. Spark Connect's `spark.sql` then executes the command eagerly on the server. So in the default read-only mode, execute_sql, explain_query and analyze_query write to persistent tables without an error. A CTE-based write is an ordinary statement that an LLM produces without adversarial intent, so the best-effort caveat does not cover it. No test sends a WITH-prefixed statement.
**Recommended change:** In read-only mode, run SELECT- and WITH-led statements as a derived table, `SELECT * FROM (<statement>) AS <alias>` with the closing parenthesis on its own line, so that Spark's parser rejects anything that is not a query. Keep the lead-keyword check for the other allow-listed statement types.
**Why this works:** A derived table can contain only a query. CTE-prefixed INSERT, DELETE, UPDATE or MERGE, and any statement disguised behind a SELECT or WITH lead, therefore fails to parse instead of executing, and the rejection keeps Spark's parse message.
**Scope:** Make the read-only gate wrap query-led statements so that Spark's parser enforces query-only execution.
**Compatibility:** SHOW, DESCRIBE/DESC, EXPLAIN, USE, SET and RESET remain accepted in read-only mode, and --no-read-only still allows arbitrary SQL.
**Risks:** The derived table must preserve the statement's ordering, LIMIT and OFFSET, and its parameter markers. An ORDER BY inside a derived table under a projection-only outer SELECT should be kept, and tests need to confirm it.
**Constraints:** Strip trailing semicolons and put the closing parenthesis on its own line, so that a trailing `--` comment or `;` cannot break or escape the wrapper. When the wrapped statement fails to parse, reject it with an error that says read-only mode accepts only a single query and that carries Spark's parse message, so that a disguised write and a typo in a SELECT are both reported accurately.
**Exclusions:**
- This repair does not decide whether read-only enforcement should move to the server or should also cover crafted comments behind the other allow-listed statement types; that is the open read-only scope question.
**Success:** In read-only mode, WITH ... INSERT, DELETE, UPDATE and MERGE are rejected without executing. In read-only mode, SELECT and WITH queries return the same rows, order and schema as before. Behavior with --no-read-only is unchanged.
**Verification:**
- **Regression:** CTE-prefixed INSERT, DELETE and MERGE sent to execute_sql in read-only mode are rejected, and the target table is left unchanged.
- **Regression:** A SELECT-led statement that hides another statement behind a nested comment is rejected in read-only mode.
- **Behavior:** Ordinary SELECT and WITH queries, including ones with ORDER BY plus LIMIT or OFFSET and with named or positional parameters, return the same rows and schema as without the wrapper.
- **Behavior:** In read-only mode, a SELECT with a syntax error is rejected with Spark's syntax error message rather than a bare read-only violation.
##########
python/pyspark/sql/mcp/tools/session.py:
##########
@@ -0,0 +1,79 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from __future__ import annotations
+
+from typing import Any, Dict
+
+from pyspark.sql.mcp.session import SessionHolder
+from pyspark.sql.mcp.tools.registry import ToolSpec
+
+# Configs that are safe to surface to an LLM. Anything else is filtered out
+# to avoid leaking tokens or workspace-scoped identifiers.
+_SAFE_CONFIG_PREFIXES = (
+ "spark.app.",
+ "spark.sql.",
+ "spark.connect.",
+)
+_REDACT_SUBSTRINGS = ("token", "secret", "password", "credential")
+
+
+async def _handle(_args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ spark = holder.get()
+ config = holder.config
+
+ version = spark.version
+ current_catalog = spark.catalog.currentCatalog()
+ current_database = spark.catalog.currentDatabase()
+ safe_configs = _collect_safe_configs(spark)
+
+ return {
+ "spark_version": version,
+ "connect_url": config.connect_url,
Review Comment:
**Blocking (P1):** `config.connect_url` is returned verbatim, and a Spark Connect URL carries its bearer token as a parameter (`ChannelBuilder` reads `token` from it), e.g. `sc://host:443/;use_ssl=true;token=<secret>`. On every token-authenticated deployment, the first `get_session_info` call (its description says to call it first) puts the credential into the LLM's context, its transcripts and provider-side logs. The config filter in this same tool exists to prevent exactly that.
See **Shared repair plan 2** in the review body.
##########
python/pyspark/sql/mcp/session.py:
##########
@@ -0,0 +1,58 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from typing import TYPE_CHECKING, Optional
+
+from pyspark.sql.mcp.config import ServerConfig
+
+if TYPE_CHECKING:
+ from pyspark.sql.session import SparkSession
+
+
+class SessionHolder:
+ """Lazily creates and caches a Spark Connect session for the MCP server.
+
+ One MCP client connection maps to one Spark Connect session.
+ """
+
+ def __init__(self, config: ServerConfig):
+ self._config = config
+ self._session: Optional["SparkSession"] = None
+
+ @property
+ def config(self) -> ServerConfig:
+ return self._config
+
+ def get(self) -> "SparkSession":
+ if self._session is None:
+ from pyspark.sql import SparkSession
+
+ builder = SparkSession.builder.remote(self._config.connect_url)
+ for key, value in self._config.extra_session_configs.items():
+ builder = builder.config(key, value)
+ if self._config.user_id is not None:
+ builder = builder.config("spark.connect.session.userId", self._config.user_id)
+ self._session = builder.getOrCreate()
Review Comment:
**Non-blocking (P2):** `getOrCreate()` returns any active or default Connect session already in the process, ignoring `spark.remote`, and then re-applies this builder's options to it as runtime confs. `build_server` is exported, so embedding it in a process that already holds a session (a notebook, a second server) silently runs every tool against that other session, while `get_session_info` reports the configured URL. It also pushes the user-id key and the extra configs onto the caller's session.
See **Shared repair plan 3** in the review body.
##########
python/pyspark/sql/mcp/tools/session.py:
##########
@@ -0,0 +1,79 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from __future__ import annotations
+
+from typing import Any, Dict
+
+from pyspark.sql.mcp.session import SessionHolder
+from pyspark.sql.mcp.tools.registry import ToolSpec
+
+# Configs that are safe to surface to an LLM. Anything else is filtered out
+# to avoid leaking tokens or workspace-scoped identifiers.
+_SAFE_CONFIG_PREFIXES = (
+ "spark.app.",
+ "spark.sql.",
+ "spark.connect.",
+)
+_REDACT_SUBSTRINGS = ("token", "secret", "password", "credential")
+
+
+async def _handle(_args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ spark = holder.get()
+ config = holder.config
+
+ version = spark.version
+ current_catalog = spark.catalog.currentCatalog()
+ current_database = spark.catalog.currentDatabase()
+ safe_configs = _collect_safe_configs(spark)
+
+ return {
+ "spark_version": version,
+ "connect_url": config.connect_url,
+ "read_only": config.read_only,
+ "current_catalog": current_catalog,
+ "current_database": current_database,
+ "configs": safe_configs,
+ }
+
+
+def _collect_safe_configs(spark: Any) -> Dict[str, str]:
+ try:
+ items = spark.conf.getAll()
Review Comment:
**Non-blocking (P2):** `getAll` is a `@property` on both the classic and the Connect `RuntimeConf`, so `spark.conf.getAll()` calls the returned dict and raises `TypeError: 'dict' object is not callable`, which the bare `except` turns into `{}`. Against any real session, `configs` is therefore always empty. The unit fake's `getAll` is a plain method, so the test does not notice.
See **Shared repair plan 2** in the review body.
##########
python/pyspark/sql/mcp/config.py:
##########
@@ -0,0 +1,100 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import os
+from dataclasses import dataclass, field
+from typing import Optional, TypeVar
+
+from pyspark.errors import PySparkValueError
+
+
+DEFAULT_MAX_ROWS = 1000
+DEFAULT_PAGE_SIZE = 100
+DEFAULT_QUERY_TIMEOUT_SECONDS = 60
+
+
+@dataclass(frozen=True)
+class ServerConfig:
+ """Runtime configuration for the Spark MCP server.
+
+ Use :meth:`from_env` to populate from CLI args + environment variables.
+ """
+
+ connect_url: str
+ read_only: bool = True
+ max_rows: int = DEFAULT_MAX_ROWS
+ default_page_size: int = DEFAULT_PAGE_SIZE
+ query_timeout_seconds: int = DEFAULT_QUERY_TIMEOUT_SECONDS
+ transport: str = "stdio"
+ user_id: Optional[str] = None
+ extra_session_configs: dict = field(default_factory=dict)
+
+ @classmethod
+ def from_env(
+ cls,
+ *,
+ connect_url: Optional[str] = None,
+ read_only: Optional[bool] = None,
+ transport: Optional[str] = None,
+ max_rows: Optional[int] = None,
+ query_timeout_seconds: Optional[int] = None,
+ user_id: Optional[str] = None,
+ ) -> "ServerConfig":
+ url = connect_url or os.environ.get("SPARK_REMOTE")
+ if not url:
+ raise PySparkValueError(
+ errorClass="MCP_CONNECT_URL_REQUIRED",
+ messageParameters={},
+ )
+ return cls(
+ connect_url=url,
+ read_only=_default(read_only, _env_bool("SPARK_MCP_READ_ONLY", True)),
+ max_rows=_default(max_rows, _env_int("SPARK_MCP_MAX_ROWS", DEFAULT_MAX_ROWS)),
Review Comment:
**Nit (P3):** `_env_int(...)` is evaluated before `_default` sees the CLI value, so `SPARK_MCP_MAX_ROWS=1k python -m pyspark.sql.mcp --max-rows 500` fails with MCP_INVALID_ENV_INT even though the flag would win. The timeout variable behaves the same way. This contradicts the README's 'CLI flags take precedence over env vars'. Reading the variable only when the CLI value is None fixes it.
**Verification:**
- **Regression:** An invalid SPARK_MCP_MAX_ROWS combined with a valid --max-rows starts with the CLI value and no error.
- **Behavior:** An invalid environment value without a CLI override still fails with the integer error.
##########
python/pyspark/sql/mcp/tools/session.py:
##########
@@ -0,0 +1,79 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from __future__ import annotations
+
+from typing import Any, Dict
+
+from pyspark.sql.mcp.session import SessionHolder
+from pyspark.sql.mcp.tools.registry import ToolSpec
+
+# Configs that are safe to surface to an LLM. Anything else is filtered out
+# to avoid leaking tokens or workspace-scoped identifiers.
+_SAFE_CONFIG_PREFIXES = (
+ "spark.app.",
+ "spark.sql.",
+ "spark.connect.",
+)
+_REDACT_SUBSTRINGS = ("token", "secret", "password", "credential")
Review Comment:
**Non-blocking (P2):** This list checks key names only. It misses Spark's own `access[.]?key` pattern (`spark.redaction.regex` defaults to `(?i)secret|password|token|access[.]?key`) and never checks values, which Connect's GetAll returns unredacted. The filter is dormant only because `getAll()` always raises today (see the always-empty configs finding). Once that call works, `spark.sql.catalog.<c>.fs.s3a.access.key`, or a catalog URL value containing `password=...`, reaches the LLM. Redacting both keys and values with Spark's pattern avoids that.
See **Shared repair plan 2** in the review body.
##########
python/pyspark/sql/mcp/tools/query.py:
##########
@@ -0,0 +1,363 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""SQL execution and plan tools."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Awaitable, Callable, Dict, List, Optional, TypeVar
+
+from pyspark.errors import PySparkRuntimeError, PySparkValueError
+from pyspark.sql.mcp.safety import assert_read_only
+from pyspark.sql.mcp.session import SessionHolder
+from pyspark.sql.mcp.tools.catalog import _to_dict
+from pyspark.sql.mcp.tools.registry import ToolSpec
+
+
+T = TypeVar("T")
+
+
+async def _with_timeout(holder: SessionHolder, func: Callable[[], T], *, label: str) -> T:
+ """Run a blocking Spark Connect call in a worker thread under the configured timeout."""
+ timeout = holder.config.query_timeout_seconds
+ coro: Awaitable[T] = asyncio.to_thread(func)
+ if timeout and timeout > 0:
+ try:
+ return await asyncio.wait_for(coro, timeout=timeout)
Review Comment:
**Non-blocking (P2):** On expiry, `wait_for` only cancels the awaiting coroutine. The `to_thread` worker keeps running `collect` or `_explain_string`, and the ExecutePlan or AnalyzePlan continues on the server, because nothing tags or interrupts it (Connect provides `addTag` and `interruptTag` for this). Running this exact pattern locally, the timeout fired at 0.3 s while the worker ran until 1.2 s, and `asyncio.run` returned only then. Timed-out queries and LLM retries therefore keep cluster work running and hold threads of the default executor (`min(32, cpu+4)`), so later calls can queue behind them.
This extends an [existing discussion](https://github.com/apache/spark/pull/55648#issuecomment-4377110887).
See **Shared repair plan 1** in the review body.
##########
python/pyspark/sql/mcp/server.py:
##########
@@ -0,0 +1,175 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Entry point for the Spark MCP server.
+
+The :func:`build_server` function is the seam used by tests; :func:`run` is
+the CLI entry point used by ``python -m pyspark.sql.mcp``.
+
+The official ``mcp`` Python SDK is imported lazily so that importing this
+module does not pull the SDK in on machines that only need the configuration
+objects (e.g. for unit testing the safety layer).
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from typing import Any, TYPE_CHECKING
+
+from pyspark.errors import PySparkNotImplementedError, PySparkValueError
+from pyspark.sql.mcp.config import ServerConfig
+from pyspark.sql.mcp.session import SessionHolder
+from pyspark.sql.mcp.tools.registry import all_tools
+
+if TYPE_CHECKING:
+ from mcp.server import Server
+
+
+def build_server(config: ServerConfig) -> "Server":
+ """Build an MCP server instance bound to a Spark Connect session.
+
+ The server registers tool handlers but does not start a transport. Use
+ :func:`run` (or the SDK's transport runners) to serve.
+ """
+ from mcp.server import Server
+
+ holder = SessionHolder(config)
+ server: "Server[Any]" = Server("spark-mcp")
+ _register_tools(server, holder)
+ return server
+
+
+def _register_tools(server: "Server", holder: SessionHolder) -> None:
+ import mcp.types as types
+
+ specs = {spec.name: spec for spec in all_tools()}
+
+ @server.list_tools()
+ async def _list_tools() -> list:
+ return [
+ types.Tool(
+ name=spec.name,
+ description=spec.description,
+ inputSchema=spec.input_schema,
+ )
+ for spec in specs.values()
+ ]
+
+ @server.call_tool()
+ async def _call_tool(name: str, arguments: dict) -> list:
+ spec = specs.get(name)
+ if spec is None:
+ raise PySparkValueError(
+ errorClass="MCP_UNKNOWN_TOOL",
+ messageParameters={"name": name},
+ )
+ result = await spec.handler(arguments or {}, holder)
+ return [types.TextContent(type="text", text=json.dumps(result, default=str))]
Review Comment:
**Nit (P3):** `default=str` applies only to values: `json.dumps` raises `TypeError: keys must be str, int, float, bool or None` for a dict key that is a date, a Decimal or a tuple (checked for all three), and `Row.asDict(recursive=True)` leaves map keys as they are. `SELECT map(DATE'2024-01-01', 1) AS m`, or any map column keyed by date, timestamp, decimal, binary or struct values, therefore returns a serialization error instead of rows in the default JSON format. Stringifying such keys during row conversion fixes it.
**Verification:**
- **Regression:** A row containing a map keyed by a date serializes to JSON, with the key rendered as a string.
##########
python/pyspark/sql/mcp/README.md:
##########
@@ -0,0 +1,186 @@
+# Spark MCP Server
+
+A [Model Context Protocol](https://modelcontextprotocol.io) server for Apache
+Spark, implemented as a thin client over [Spark
+Connect](https://spark.apache.org/docs/latest/spark-connect-overview.html).
+
+## Why
+
+LLM clients can already talk to MCP servers; Spark Connect already separates
+client from cluster. This module connects the two: a Spark cluster shows up
+to an LLM as a set of safe, paginated tools — `list_tables`, `describe_table`,
+`execute_sql`, `explain_query`, etc. — without giving the LLM a JVM driver.
+
+## Run
+
+The server speaks the `stdio` MCP transport (other transports may be added
+later) and connects to a Spark cluster via Spark Connect:
+
+```bash
+python -m pyspark.sql.mcp \
+ --connect-url "sc://localhost:15002"
+```
+
+Equivalent via environment variable:
+
+```bash
+export SPARK_REMOTE="sc://localhost:15002"
+python -m pyspark.sql.mcp
+```
+
+The server is **read-only by default**. Pass `--no-read-only` to allow DDL/DML
+(use with care; the read-only filter is a guardrail, not a security boundary).
+
+## Configuring an MCP client
+
+The server speaks `stdio` MCP transport, so a client launches it as a
+subprocess. The exact configuration step is client-specific; the moving
+parts are always:
+
+1. The command (`python -m pyspark.sql.mcp`).
+2. The Spark Connect URL, supplied via `--connect-url` or the
+ `SPARK_REMOTE` environment variable.
+3. Optionally, `--no-read-only` to permit DDL/DML.
+
+### Claude Code
+
+```bash
+claude mcp add spark \
+ -e SPARK_REMOTE=sc://localhost:15002 \
+ -- python -m pyspark.sql.mcp
+```
+
+Pass `--scope user` to register the server for all projects, or
+`--scope project` to write a shared `.mcp.json` into the current
+repository. Restart the Claude Code session and the LLM gains
+`mcp__spark__*` tools.
+
+### Claude Desktop
+
+Edit the application's `claude_desktop_config.json` (location is
+documented in the Claude Desktop docs) and add an entry under
+`mcpServers`:
+
+```json
+{
+ "mcpServers": {
+ "spark": {
+ "command": "python",
+ "args": ["-m", "pyspark.sql.mcp"],
+ "env": {
+ "SPARK_REMOTE": "sc://localhost:15002"
+ }
+ }
+ }
+}
+```
+
+Restart Claude Desktop.
+
+### Other MCP clients
+
+Any MCP-compliant client that can launch a stdio subprocess can use the
+same shape:
+
+```json
+{
+ "command": "python",
+ "args": ["-m", "pyspark.sql.mcp"],
+ "env": {"SPARK_REMOTE": "sc://localhost:15002"}
+}
+```
+
+### Environment variables
+
+| Variable | Meaning |
+| --- | --- |
+| `SPARK_REMOTE` | Spark Connect URL (`sc://host:port[/;params]`). |
+| `SPARK_MCP_READ_ONLY` | `true` (default) or `false`. |
+| `SPARK_MCP_USER_ID` | Optional Spark Connect session user id. |
+| `SPARK_MCP_MAX_ROWS` | Hard cap on rows returned per tool call (default `1000`). |
+| `SPARK_MCP_QUERY_TIMEOUT_SECONDS` | Wall-clock cap per tool call (default `60`; `0` disables). |
+| `SPARK_MCP_TRANSPORT` | Reserved; only `stdio` is implemented. |
+
+Equivalent CLI flags: `--user-id`, `--max-rows`, `--query-timeout-seconds`.
+CLI flags take precedence over env vars.
+
+To verify the wiring, ask the LLM to list catalogs, or to call
+`get_session_info` — the response should include the Spark version
+reported by the cluster.
+
+## Tools
+
+Catalog exploration, query execution, and query plans:
+
+- `get_session_info`
+- `list_catalogs`, `list_databases`, `list_tables`, `describe_table`,
+ `list_functions`
+- `execute_sql`, `preview_table`
+- `explain_query`, `analyze_query`
+
+## Motivating example: LLM-assisted plan analysis
+
+`explain_query` and `analyze_query` are the highest-leverage tools the
+server exposes: an LLM that can read a structured Spark plan can reason
+about performance the way a Spark engineer would, without running the
+query.
+
+A user prompts an LLM client (which has the Spark MCP server attached):
+
+> "I want to understand the execution plan for
+> `SELECT count(*) FROM range(1000000) WHERE id % 7 = 0 GROUP BY id % 3` —
+> but **don't actually execute it**. From the plan, where is the predicate
+> handled?"
+
+The LLM calls `analyze_query`, receives the optimized logical and physical
+plans, and produces an answer like:
+
+```
+Optimized Logical Plan
+Aggregate [_groupingexpression#50L], [count(1) AS count(1)]
++- Project [(id % 3) AS _groupingexpression]
+ +- Filter ((id % 7) = 0)
+ +- Range (0, 1000000, step=1)
+
+Physical Plan
+AdaptiveSparkPlan isFinalPlan=false
++- HashAggregate(keys=[_groupingexpression], functions=[count(1)])
+ +- Exchange hashpartitioning(_groupingexpression, 200)
+ +- HashAggregate(keys=[_groupingexpression], functions=[partial_count(1)])
+ +- Project [(id % 3) AS _groupingexpression]
+ +- Filter ((id % 7) = 0)
+ +- Range (0, 1000000, step=1, splits=16)
+```
+
+> 1. **Range (splits=16)** generates 0..999999 across 16 partitions — no I/O.
+> 2. **Filter `(id % 7) = 0`** sits directly above the Range, so only ~1/7 of
+> the rows survive; everything above it sees the smaller stream.
+> 3. **Project `id % 3`** materializes the group key once so it isn't
+> recomputed on either side of the aggregate.
+> 4. **HashAggregate (partial_count)** does map-side combine inside each
+> partition, emitting `(group_key, partial_count)`.
+> 5. **Exchange hashpartitioning(200)** shuffles by group key. Since the key
+> only has 3 distinct values, only 3 reducers actually receive data.
+> 6. **HashAggregate (final count)** sums the partial counts into the final
+> 3-row result.
+> 7. **AdaptiveSparkPlan `isFinalPlan=false`** means AQE can still adjust at
+> runtime (coalesce empty partitions, change join strategies, etc.).
+>
+> **Where is `id % 7 = 0` handled?** In the `Filter` operator immediately
+> above the source — the lowest, earliest point in the plan. The
+> `PushDownPredicates` optimizer rule moves filters as close to the source
+> as possible. `Range` isn't a file source, so there is no storage-level
+> pushdown, but the effect is identical: `id % 3`, the partial aggregate,
+> and the shuffle all see only the post-filter rows. For a Parquet source
+> the same filter would also be pushed down as a data-source filter and
Review Comment:
**Nit (P3):** For `(id % 7) = 0`, this is not what Spark does. Data-source filter translation accepts only comparisons on a column or a nested field (`PushableColumn` in DataSourceStrategy), so a predicate on `Remainder(id, 7)` is not pushed to Parquet as a data filter; only `IsNotNull(id)` would be. The filter could prune partitions only if `id` were a partition column, which the example does not state. Because the README presents this answer as the kind of explanation that makes the plan tools valuable, it teaches users, and the LLMs it primes, a wrong rule.
**Verification:**
- **Inspection:** The pushdown statements in the example match how Spark translates filters for data sources.
##########
python/pyspark/sql/mcp/server.py:
##########
@@ -0,0 +1,175 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Entry point for the Spark MCP server.
+
+The :func:`build_server` function is the seam used by tests; :func:`run` is
Review Comment:
**Nit (P3):** No test calls `build_server`; both test modules go through `all_tools()`, so it is not the seam used by tests. The `build_server` docstring below also says to serve the returned instance with `run`, but `run(argv)` parses CLI arguments and builds its own server, and only the private `_serve` can serve an existing instance. Both sentences should describe the actual usage.
**Verification:**
- **Inspection:** Each usage claim in the server docstrings matches the code and the tests.
##########
python/pyspark/sql/mcp/server.py:
##########
@@ -0,0 +1,175 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Entry point for the Spark MCP server.
+
+The :func:`build_server` function is the seam used by tests; :func:`run` is
+the CLI entry point used by ``python -m pyspark.sql.mcp``.
+
+The official ``mcp`` Python SDK is imported lazily so that importing this
+module does not pull the SDK in on machines that only need the configuration
+objects (e.g. for unit testing the safety layer).
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from typing import Any, TYPE_CHECKING
+
+from pyspark.errors import PySparkNotImplementedError, PySparkValueError
+from pyspark.sql.mcp.config import ServerConfig
+from pyspark.sql.mcp.session import SessionHolder
+from pyspark.sql.mcp.tools.registry import all_tools
+
+if TYPE_CHECKING:
+ from mcp.server import Server
+
+
+def build_server(config: ServerConfig) -> "Server":
+ """Build an MCP server instance bound to a Spark Connect session.
+
+ The server registers tool handlers but does not start a transport. Use
+ :func:`run` (or the SDK's transport runners) to serve.
+ """
+ from mcp.server import Server
+
+ holder = SessionHolder(config)
Review Comment:
**Non-blocking (P2):** This holder is reachable only from the tool closures, so neither `run` nor `_serve` can close it, and `SessionHolder.close` has no caller. The Connect client sends ReleaseSession at exit only when `SPARK_CONNECT_RELEASE_SESSION_ON_EXIT` is set, and it defaults to false. Each server shutdown, which happens on every MCP client restart, therefore leaves a server-side session with its temp views, confs and cached state until the idle-session timeout expires.
See **Shared repair plan 3** in the review body.
##########
python/pyspark/sql/mcp/session.py:
##########
@@ -0,0 +1,58 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from typing import TYPE_CHECKING, Optional
+
+from pyspark.sql.mcp.config import ServerConfig
+
+if TYPE_CHECKING:
+ from pyspark.sql.session import SparkSession
+
+
+class SessionHolder:
+ """Lazily creates and caches a Spark Connect session for the MCP server.
+
+ One MCP client connection maps to one Spark Connect session.
+ """
+
+ def __init__(self, config: ServerConfig):
+ self._config = config
+ self._session: Optional["SparkSession"] = None
+
+ @property
+ def config(self) -> ServerConfig:
+ return self._config
+
+ def get(self) -> "SparkSession":
+ if self._session is None:
+ from pyspark.sql import SparkSession
+
+ builder = SparkSession.builder.remote(self._config.connect_url)
+ for key, value in self._config.extra_session_configs.items():
+ builder = builder.config(key, value)
+ if self._config.user_id is not None:
+ builder = builder.config("spark.connect.session.userId", self._config.user_id)
Review Comment:
**Non-blocking (P2):** Nothing reads `spark.connect.session.userId`. The Connect builder constructs the client without a user id and only afterwards applies leftover builder options as silent session confs. The client takes its user from the URL's `user_id` parameter, the `userId` constructor argument, or `SPARK_USER`/`USER`. A server started with `--user-id` or `SPARK_MCP_USER_ID` therefore still runs as the OS user, and the per-user isolation and attribution the operator configured never apply.
See **Shared repair plan 3** in the review body.
##########
python/pyspark/sql/mcp/tools/catalog.py:
##########
@@ -0,0 +1,218 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Catalog browsing tools."""
+
+from __future__ import annotations
+
+from dataclasses import asdict, is_dataclass
+from typing import Any, Dict, List
+
+from pyspark.sql.mcp.session import SessionHolder
+from pyspark.sql.mcp.tools.registry import ToolSpec
+
+
+# ---------------------------------------------------------------------------
+# Result shaping
+# ---------------------------------------------------------------------------
+
+
+def _to_dict(obj: Any) -> Dict[str, Any]:
+ """Best-effort conversion of catalog dataclasses (CatalogMetadata, Database,
+ Table, Column, Function) into plain dicts."""
+ if is_dataclass(obj) and not isinstance(obj, type):
+ return asdict(obj)
+ if hasattr(obj, "_asdict"): # NamedTuple
+ return dict(obj._asdict())
+ if isinstance(obj, dict):
+ return obj
+ # Fallback: scrape public attributes.
+ return {
+ k: getattr(obj, k)
+ for k in dir(obj)
+ if not k.startswith("_") and not callable(getattr(obj, k))
+ }
+
+
+def _paginate(items: List[Any], limit: int, offset: int) -> Dict[str, Any]:
+ total = len(items)
+ end = offset + limit if limit is not None else total
+ page = items[offset:end]
+ return {
+ "items": [_to_dict(x) for x in page],
+ "offset": offset,
+ "limit": limit,
+ "returned": len(page),
+ "total": total,
+ "truncated": end < total,
+ }
+
+
+# ---------------------------------------------------------------------------
+# list_catalogs
+# ---------------------------------------------------------------------------
+
+
+async def _handle_list_catalogs(args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ spark = holder.get()
+ catalogs = list(spark.catalog.listCatalogs())
+ limit = int(args.get("limit", holder.config.default_page_size))
+ offset = int(args.get("offset", 0))
+ return _paginate(catalogs, limit, offset)
+
+
+def list_catalogs_spec() -> ToolSpec:
+ return ToolSpec(
+ name="list_catalogs",
+ description="List catalogs available to the current Spark session.",
+ input_schema={
+ "type": "object",
+ "properties": {
+ "limit": {"type": "integer", "minimum": 1, "maximum": 1000, "default": 100},
+ "offset": {"type": "integer", "minimum": 0, "default": 0},
+ },
+ "additionalProperties": False,
+ },
+ handler=_handle_list_catalogs,
+ )
+
+
+# ---------------------------------------------------------------------------
+# list_databases
+# ---------------------------------------------------------------------------
+
+
+async def _handle_list_databases(args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ spark = holder.get()
+ catalog = args.get("catalog")
+ pattern = args.get("pattern")
+
+ if catalog:
+ spark.catalog.setCurrentCatalog(catalog)
Review Comment:
**Non-blocking (P2):** `setCurrentCatalog` is never undone, and switching catalogs also resets the current namespace and the v1 current database to `default` (CatalogManager.setCurrentCatalog). After a read-style `list_databases(catalog='other')`, unqualified names in execute_sql, preview_table, describe_table and the plan tools resolve in `other`'s default namespace, and `get_session_info` reports the switched catalog. The MCP SDK runs requests concurrently, so a call that is already mid-collect can be affected too. A repro on mcp 1.30.0 showed the switch landing between execute_sql's schema read and its collect, and the collect re-analyzes the SQL under the new catalog.
**Recommended change:** When a catalog is given, run `SHOW NAMESPACES IN <backtick-quoted catalog>`, with `LIKE <pattern>` when a pattern is given, and return each namespace with its catalog. Keep spark.catalog.listDatabases for the current catalog.
**Why this works:** The named-catalog listing no longer touches the current catalog or namespace, so later and concurrent calls resolve names exactly as before.
**Scope:** List a named catalog's databases without changing the session's current catalog.
**Compatibility:** list_databases without a catalog returns the same items as before.
**Risks:** Items for a named catalog carry the name and the catalog, but not the description or location, which SHOW NAMESPACES does not return.
**Constraints:** Quote the catalog name, escaping any backticks, and pass the pattern through unchanged.
**Success:** After listing another catalog's databases, the session's current catalog and current database are unchanged. The listing contains that catalog's namespaces, filtered by the pattern.
**Verification:**
- **Regression:** Listing the databases of another catalog leaves the reported current catalog and database unchanged, and unqualified queries still resolve in the original namespace.
- **Behavior:** A named-catalog listing returns the same namespace names as SHOW NAMESPACES IN that catalog, with the pattern applied.
##########
python/pyspark/sql/mcp/config.py:
##########
@@ -0,0 +1,100 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import os
+from dataclasses import dataclass, field
+from typing import Optional, TypeVar
+
+from pyspark.errors import PySparkValueError
+
+
+DEFAULT_MAX_ROWS = 1000
+DEFAULT_PAGE_SIZE = 100
+DEFAULT_QUERY_TIMEOUT_SECONDS = 60
+
+
+@dataclass(frozen=True)
+class ServerConfig:
+ """Runtime configuration for the Spark MCP server.
+
+ Use :meth:`from_env` to populate from CLI args + environment variables.
+ """
+
+ connect_url: str
+ read_only: bool = True
+ max_rows: int = DEFAULT_MAX_ROWS
+ default_page_size: int = DEFAULT_PAGE_SIZE
+ query_timeout_seconds: int = DEFAULT_QUERY_TIMEOUT_SECONDS
+ transport: str = "stdio"
+ user_id: Optional[str] = None
+ extra_session_configs: dict = field(default_factory=dict)
+
+ @classmethod
+ def from_env(
+ cls,
+ *,
+ connect_url: Optional[str] = None,
+ read_only: Optional[bool] = None,
+ transport: Optional[str] = None,
+ max_rows: Optional[int] = None,
+ query_timeout_seconds: Optional[int] = None,
+ user_id: Optional[str] = None,
+ ) -> "ServerConfig":
+ url = connect_url or os.environ.get("SPARK_REMOTE")
+ if not url:
+ raise PySparkValueError(
+ errorClass="MCP_CONNECT_URL_REQUIRED",
+ messageParameters={},
+ )
+ return cls(
+ connect_url=url,
+ read_only=_default(read_only, _env_bool("SPARK_MCP_READ_ONLY", True)),
+ max_rows=_default(max_rows, _env_int("SPARK_MCP_MAX_ROWS", DEFAULT_MAX_ROWS)),
+ query_timeout_seconds=_default(
+ query_timeout_seconds,
+ _env_int("SPARK_MCP_QUERY_TIMEOUT_SECONDS", DEFAULT_QUERY_TIMEOUT_SECONDS),
+ ),
+ transport=transport or os.environ.get("SPARK_MCP_TRANSPORT", "stdio"),
+ user_id=user_id or os.environ.get("SPARK_MCP_USER_ID"),
+ )
+
+
+_T = TypeVar("_T")
+
+
+def _default(value: Optional[_T], fallback: _T) -> _T:
+ return fallback if value is None else value
+
+
+def _env_bool(name: str, default: bool) -> bool:
+ raw = os.environ.get(name)
+ if raw is None:
+ return default
+ return raw.strip().lower() in {"1", "true", "yes", "on"}
Review Comment:
**Non-blocking (P2):** Any set value outside `{1, true, yes, on}` maps to False here. `SPARK_MCP_READ_ONLY=''`, which is common in templated MCP client env blocks, or a typo like `ture` therefore silently disables read-only mode, with no error. For a safety default that is the wrong direction; `_env_int` below already treats a blank value as unset. Keeping the default for blank values and rejecting unknown ones, as `_env_int` does, closes the gap.
**Verification:**
- **Regression:** SPARK_MCP_READ_ONLY set to an empty string, to a misspelling or to an unknown word never yields read-only off.
- **Behavior:** The values false, 0, no and off disable read-only, and true, 1, yes and on enable it.
##########
python/pyspark/sql/mcp/tools/query.py:
##########
@@ -0,0 +1,363 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""SQL execution and plan tools."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Awaitable, Callable, Dict, List, Optional, TypeVar
+
+from pyspark.errors import PySparkRuntimeError, PySparkValueError
+from pyspark.sql.mcp.safety import assert_read_only
+from pyspark.sql.mcp.session import SessionHolder
+from pyspark.sql.mcp.tools.catalog import _to_dict
+from pyspark.sql.mcp.tools.registry import ToolSpec
+
+
+T = TypeVar("T")
+
+
+async def _with_timeout(holder: SessionHolder, func: Callable[[], T], *, label: str) -> T:
+ """Run a blocking Spark Connect call in a worker thread under the configured timeout."""
+ timeout = holder.config.query_timeout_seconds
+ coro: Awaitable[T] = asyncio.to_thread(func)
+ if timeout and timeout > 0:
+ try:
+ return await asyncio.wait_for(coro, timeout=timeout)
+ except asyncio.TimeoutError as exc:
+ raise PySparkRuntimeError(
+ errorClass="MCP_QUERY_TIMEOUT",
+ messageParameters={"label": label, "timeout": str(timeout)},
+ ) from exc
+ return await coro
+
+
+# ---------------------------------------------------------------------------
+# list_functions
+# ---------------------------------------------------------------------------
+
+
+async def _handle_list_functions(args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ spark = holder.get()
+ database = args.get("database")
+ pattern = args.get("pattern")
+
+ kwargs: Dict[str, Any] = {}
+ if database is not None:
+ kwargs["dbName"] = database
+ if pattern is not None:
+ kwargs["pattern"] = pattern
+ funcs = list(spark.catalog.listFunctions(**kwargs))
+
+ limit = int(args.get("limit", holder.config.default_page_size))
+ offset = int(args.get("offset", 0))
+ total = len(funcs)
+ page = funcs[offset : offset + limit]
+ return {
+ "items": [_to_dict(f) for f in page],
+ "offset": offset,
+ "limit": limit,
+ "returned": len(page),
+ "total": total,
+ "truncated": offset + limit < total,
+ }
+
+
+def list_functions_spec() -> ToolSpec:
+ return ToolSpec(
+ name="list_functions",
+ description=(
+ "List user-defined and built-in functions visible in a database. "
+ "Useful when authoring SQL — surfaces both UDFs and the built-in "
+ "function library."
+ ),
+ input_schema={
+ "type": "object",
+ "properties": {
+ "database": {"type": "string"},
+ "pattern": {"type": "string"},
+ "limit": {"type": "integer", "minimum": 1, "maximum": 1000, "default": 100},
+ "offset": {"type": "integer", "minimum": 0, "default": 0},
+ },
+ "additionalProperties": False,
+ },
+ handler=_handle_list_functions,
+ )
+
+
+# ---------------------------------------------------------------------------
+# Result rendering helpers
+# ---------------------------------------------------------------------------
+
+
+def _row_to_dict(row: Any) -> Dict[str, Any]:
+ if hasattr(row, "asDict"):
+ return row.asDict(recursive=True)
+ if hasattr(row, "_asdict"):
+ return dict(row._asdict())
+ return dict(row)
+
+
+def _render_markdown(rows: List[Dict[str, Any]], schema_fields: List[str]) -> str:
+ if not schema_fields:
+ return "_(no columns)_"
+ header = "| " + " | ".join(schema_fields) + " |"
+ sep = "| " + " | ".join("---" for _ in schema_fields) + " |"
+ body = []
+ for row in rows:
+ body.append("| " + " | ".join(_md_cell(row.get(c)) for c in schema_fields) + " |")
+ return "\n".join([header, sep, *body])
+
+
+def _md_cell(value: Any) -> str:
+ if value is None:
+ return ""
+ text = str(value).replace("|", "\\|").replace("\n", " ")
+ return text
+
+
+def _schema_fields(schema: Any) -> List[str]:
+ try:
+ return [field.name for field in schema.fields]
+ except AttributeError:
+ return []
+
+
+# ---------------------------------------------------------------------------
+# execute_sql
+# ---------------------------------------------------------------------------
+
+
+async def _handle_execute_sql(args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ spark = holder.get()
+ query: str = args["query"]
+ sql_args: Optional[Any] = args.get("args")
+ fmt: str = args.get("format", "json")
+ limit_arg = args.get("limit", holder.config.default_page_size)
+ offset = int(args.get("offset", 0))
+
+ if holder.config.read_only:
+ assert_read_only(query)
+
+ capped_limit = min(int(limit_arg), int(holder.config.max_rows))
+ df = _sql(spark, query, sql_args)
+ schema_fields = _schema_fields(df.schema)
+
+ page_df = df.offset(offset).limit(capped_limit + 1) if offset else df.limit(capped_limit + 1)
+ collected = await _with_timeout(holder, page_df.collect, label="execute_sql")
+ rows = [_row_to_dict(r) for r in collected]
+ truncated = len(rows) > capped_limit
+ rows = rows[:capped_limit]
+
+ payload: Dict[str, Any] = {
+ "row_count": len(rows),
+ "truncated": truncated,
+ "limit": capped_limit,
+ "offset": offset,
+ "schema": [
+ {"name": field.name, "type": str(field.dataType), "nullable": field.nullable}
+ for field in df.schema.fields
+ ],
+ }
+ if fmt == "json":
+ payload["rows"] = rows
+ elif fmt == "markdown":
+ payload["markdown"] = _render_markdown(rows, schema_fields)
+ else:
+ raise PySparkValueError(
+ errorClass="MCP_UNSUPPORTED_FORMAT",
+ messageParameters={
+ "format": str(fmt),
+ "allowed": "json, markdown",
+ },
+ )
+ return payload
+
+
+def _sql(spark: Any, query: str, sql_args: Optional[Any]) -> Any:
+ if sql_args is None:
+ return spark.sql(query)
+ return spark.sql(query, args=sql_args)
+
+
+def execute_sql_spec() -> ToolSpec:
+ return ToolSpec(
+ name="execute_sql",
+ description=(
+ "Run a SQL query and return rows. Read-only by default — only "
+ "SELECT/WITH/SHOW/DESCRIBE/EXPLAIN/USE/SET are accepted unless "
+ "the server is started with --no-read-only. Results are paginated "
+ "and capped by the server's max_rows setting."
+ ),
+ input_schema={
+ "type": "object",
+ "properties": {
+ "query": {"type": "string", "description": "SQL statement."},
+ "args": {
+ "description": (
+ "Bind parameters. Object for named (':name'), array for positional ('?')."
+ ),
+ "oneOf": [{"type": "object"}, {"type": "array"}],
+ },
+ "limit": {"type": "integer", "minimum": 1, "default": 100},
+ "offset": {"type": "integer", "minimum": 0, "default": 0},
+ "format": {
+ "type": "string",
+ "enum": ["json", "markdown"],
+ "default": "json",
+ },
+ },
+ "required": ["query"],
+ "additionalProperties": False,
+ },
+ handler=_handle_execute_sql,
+ )
+
+
+# ---------------------------------------------------------------------------
+# preview_table
+# ---------------------------------------------------------------------------
+
+
+async def _handle_preview_table(args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ name = args["name"]
+ limit = int(args.get("limit", 20))
+ return await _handle_execute_sql(
+ {"query": f"SELECT * FROM {name}", "limit": limit, "format": args.get("format", "json")},
+ holder,
+ )
+
+
+def preview_table_spec() -> ToolSpec:
+ return ToolSpec(
+ name="preview_table",
+ description=(
+ "Convenience wrapper for SELECT * FROM <name> LIMIT <limit>. "
+ "Honors the server's max_rows cap and read-only mode."
+ ),
+ input_schema={
+ "type": "object",
+ "properties": {
+ "name": {"type": "string"},
+ "limit": {"type": "integer", "minimum": 1, "default": 20},
+ "format": {"type": "string", "enum": ["json", "markdown"], "default": "json"},
+ },
+ "required": ["name"],
+ "additionalProperties": False,
+ },
+ handler=_handle_preview_table,
+ )
+
+
+# ---------------------------------------------------------------------------
+# explain_query
+# ---------------------------------------------------------------------------
+
+
+_EXPLAIN_MODES = ("simple", "extended", "formatted", "cost", "codegen")
+
+
+async def _handle_explain_query(args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ spark = holder.get()
+ query = args["query"]
+ mode = args.get("mode", "formatted")
+ if mode not in _EXPLAIN_MODES:
+ raise PySparkValueError(
+ errorClass="MCP_INVALID_EXPLAIN_MODE",
+ messageParameters={
+ "allowed": ", ".join(_EXPLAIN_MODES),
+ "mode": str(mode),
+ },
+ )
+ if holder.config.read_only:
+ assert_read_only(query)
+
+ df = _sql(spark, query, args.get("args"))
Review Comment:
**Blocking (P1):** `_sql(spark, query, ...)` runs before `_explain_string` (here and in analyze_query). Spark Connect's `spark.sql` sends a SqlCommand that the server executes eagerly whenever the statement is a command, because `handleSqlCommand` forces `commandExecuted`. With `--no-read-only`, `explain_query({'query': 'DROP TABLE orders'})` or a DELETE therefore changes data before the plan is returned, and what gets explained is only the command's result relation. In read-only mode, SET, USE, RESET and CTE-prefixed writes take effect the same way. This breaks the tools' 'without executing it' contract and the README's 'don't actually execute it' example: an agent that explains a destructive statement before deciding whether to run it has already run it.
**Recommended change:** Have explain_query send `EXPLAIN <MODE> <statement>` through spark.sql and return the single plan string. Have analyze_query get its extended and formatted plans the same way, and compute the output schema only for statements that run as a query through the read-only gate's derived-table form, reporting an empty schema for commands.
**Why this works:** EXPLAIN runs as ExplainCommand, which plans its child without executing it, so no statement passed to the plan tools executes, in either mode.
**Scope:** Route the plan tools through EXPLAIN so that they never execute the statement they explain.
**Compatibility:** The response shapes of both tools (mode and plan; schema, extended_plan and formatted_plan) stay the same.
**Risks:** EXPLAIN output has to match the current plan text for each mode, so tests should compare the two for a query. ExplainCommand catches non-fatal planning failures and returns them as plan text, so without that check explain_query and analyze_query would report a query on a missing table as a successful plan.
**Constraints:** Map each tool mode to the corresponding EXPLAIN keyword, and keep named and positional parameter arguments working. Raise an error when EXPLAIN's output reports a planning failure (its text begins 'Error occurred during query planning:'), so that an invalid query still fails instead of returning its error as a plan.
**Success:** explain_query and analyze_query never execute the statement, with or without --no-read-only. Plans returned for queries are the same as before.
**Verification:**
- **Regression:** Explaining or analyzing DROP TABLE, DELETE and INSERT with read-only mode off leaves the table and its data unchanged.
- **Regression:** Explaining SET or USE in read-only mode leaves the session's configuration and current namespace unchanged.
- **Behavior:** Each explain mode returns the same plan text for a SELECT as before.
- **Behavior:** Explaining a query that references a missing table returns an error, as it does today.
##########
python/pyspark/sql/mcp/server.py:
##########
@@ -0,0 +1,175 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Entry point for the Spark MCP server.
+
+The :func:`build_server` function is the seam used by tests; :func:`run` is
+the CLI entry point used by ``python -m pyspark.sql.mcp``.
+
+The official ``mcp`` Python SDK is imported lazily so that importing this
+module does not pull the SDK in on machines that only need the configuration
+objects (e.g. for unit testing the safety layer).
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from typing import Any, TYPE_CHECKING
+
+from pyspark.errors import PySparkNotImplementedError, PySparkValueError
+from pyspark.sql.mcp.config import ServerConfig
+from pyspark.sql.mcp.session import SessionHolder
+from pyspark.sql.mcp.tools.registry import all_tools
+
+if TYPE_CHECKING:
+ from mcp.server import Server
+
+
+def build_server(config: ServerConfig) -> "Server":
+ """Build an MCP server instance bound to a Spark Connect session.
+
+ The server registers tool handlers but does not start a transport. Use
+ :func:`run` (or the SDK's transport runners) to serve.
+ """
+ from mcp.server import Server
+
+ holder = SessionHolder(config)
+ server: "Server[Any]" = Server("spark-mcp")
+ _register_tools(server, holder)
+ return server
+
+
+def _register_tools(server: "Server", holder: SessionHolder) -> None:
+ import mcp.types as types
+
+ specs = {spec.name: spec for spec in all_tools()}
+
+ @server.list_tools()
Review Comment:
**Non-blocking (P2):** `Server.list_tools()` and `Server.call_tool()` are the mcp 1.x registration API. mcp 2.0.0, released on 2026-07-28 after this commit, replaced them with `on_list_tools` and `on_call_tool` constructor arguments, and nothing here declares an mcp version. A repro with the current release, mcp 2.3.0, fails on this line with `AttributeError: 'Server' object has no attribute 'list_tools'`, so `python -m pyspark.sql.mcp` cannot start on a fresh install; with mcp 1.30.0 the same code builds and runs. The handlers also rely on the SDK's input validation, which starts in 1.10.0: on 1.30.0, `format='csv'`, `limit=0` and `offset=-3` were rejected before reaching Spark. Before 1.10 those arguments reach the handlers, and execute_sql runs the statement, including a write under `--no-read-only`, before it rejects the format.
**Recommended change:** State the supported range, mcp>=1.10,<2, in the README's Dependencies section and in any packaging extra or test requirement that installs mcp. In build_server, check the installed mcp release before registering tools and raise a clear PySpark error that names the supported range when it is outside it. In execute_sql, check format before running the statement.
**Why this works:** An installation inside the declared range both starts and rejects out-of-schema arguments before any statement runs. One outside it stops at startup with an error that names the range, instead of failing with AttributeError or running a statement before its arguments are rejected.
**Scope:** Declare and enforce the mcp SDK range that the adapter is written for.
**Compatibility:** Within the supported range, tool listing, dispatch and results are unchanged.
**Risks:** Users who already installed mcp 2.x must install a 1.x release to run the server until the adapter is ported. The startup check reads the installed mcp distribution's version; an installation without package metadata needs a fallback, such as checking that the low-level Server still has the list_tools and call_tool registration methods.
**Constraints:** Keep the lower bound at 1.10, the first release whose call_tool validates arguments against inputSchema, because the handlers rely on that validation for limit and offset. Pin any test requirement that installs mcp to the same range, so that adapter tests run against a release the adapter supports.
**Exclusions:**
- Porting the adapter to the mcp 2.x handler API is left to the SPIP decision about the mcp dependency.
**Success:** With mcp 2.x installed, starting the server fails immediately with an error that names the supported mcp range. With mcp 1.10 or a later 1.x release, the server starts, lists its ten tools, and rejects arguments outside each tool's input schema before running any statement. execute_sql rejects an unsupported format before it runs the statement. The README states the supported mcp range.
**Verification:**
- **Regression:** With mcp 2.x installed, building the server raises the new error naming the supported range instead of AttributeError.
- **Regression:** With a supported mcp 1.x release, execute_sql called with format 'csv', limit 0 or a negative offset returns an input validation error and runs no statement.
- **Regression:** Calling the execute_sql handler directly with an unsupported format raises the format error without calling spark.sql.
- **Behavior:** With a supported mcp 1.x release, the server starts and lists all ten tools.
##########
python/pyspark/sql/mcp/safety.py:
##########
@@ -0,0 +1,53 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import re
+
+from pyspark.errors import PySparkValueError
+
+# Read-only allow-list. Comments and leading whitespace are stripped before
+# matching. We deliberately keep this conservative — anything not on the list
Review Comment:
**Nit (P3):** This comment uses a U+2014 em-dash, and so do test_mcp_tools.py:136 and the tool-description strings at tools/query.py:86 and 202. None of them needs a non-ASCII character, and the repository guidance asks for ASCII in code and comments. Replacing each with `-` fixes all four.
**Verification:**
- **Inspection:** No non-ASCII character remains in the new Python files.
##########
python/pyspark/sql/tests/mcp/test_mcp_integration.py:
##########
@@ -0,0 +1,212 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""End-to-end integration test for the Spark MCP server.
+
+This test boots a local Spark Connect session (via the in-process server
+provided by ``SparkSession.builder.remote("local[*]")``) and exercises the
+MCP tool handlers against it. The unit tests in ``test_mcp_tools`` cover the
+same tools against a fake session; this file confirms that the handlers
+actually agree with what the real Connect client returns.
+
+The test is skipped automatically when:
+ * Spark Connect dependencies (grpcio, pandas, pyarrow, ...) are missing.
+ * Spark JARs are not built (the ``SparkSession`` cannot be created).
+
+Run with::
+
+ PYTHONPATH=python python -m unittest \
+ pyspark.sql.tests.mcp.test_mcp_integration
+"""
+
+import asyncio
+import unittest
+
+from pyspark.testing.utils import (
+ connect_requirement_message,
+ should_test_connect,
+)
+
+
+@unittest.skipIf(not should_test_connect, connect_requirement_message)
+class MCPToolsConnectIntegrationTest(unittest.TestCase):
+ """Run MCP tool handlers against a real Spark Connect session."""
+
+ spark = None
+ holder = None
+
+ @classmethod
+ def setUpClass(cls):
+ from pyspark.sql import SparkSession
+
+ cls.spark = (
+ SparkSession.builder.appName(cls.__name__)
+ # Disable the Spark UI: in some local builds Jetty's classpath
+ # is mismatched and binding the UI throws. We don't need it for
+ # an in-process test.
+ .config("spark.ui.enabled", "false")
+ # Force loopback for both driver and executor traffic so the
+ # in-process Connect REPL artifact server is reachable from
+ # local executors.
+ .config("spark.driver.bindAddress", "127.0.0.1")
+ .config("spark.driver.host", "127.0.0.1")
+ .remote("local[2]")
+ .getOrCreate()
+ )
+
+ # Seed a small dataset so the catalog/preview tools have something
+ # to chew on.
+ cls.spark.sql(
+ "CREATE TABLE IF NOT EXISTS mcp_test_orders (id BIGINT, amount DECIMAL(10,2)) USING parquet"
+ )
+ cls.spark.sql("INSERT INTO mcp_test_orders VALUES (1, 10.50), (2, 20.00), (3, 30.75)")
+
+ from pyspark.sql.mcp.config import ServerConfig
+ from pyspark.sql.mcp.session import SessionHolder
+
+ class _PreboundHolder(SessionHolder):
+ """A SessionHolder that wraps an already-built SparkSession.
+
+ The default holder calls ``SparkSession.builder.remote(...)``;
+ for the integration test we want to reuse the session created
+ in ``setUpClass`` so all assertions run against the same Spark
+ instance.
+ """
+
+ def __init__(self, config, session):
+ super().__init__(config)
+ self._session = session
+
+ cls.holder = _PreboundHolder(
+ ServerConfig(connect_url="local[2]"),
+ cls.spark,
+ )
+
+ @classmethod
+ def tearDownClass(cls):
+ try:
+ if cls.spark is not None:
+ cls.spark.sql("DROP TABLE IF EXISTS mcp_test_orders")
+ cls.spark.stop()
+ finally:
+ cls.spark = None
+ cls.holder = None
+
+ # ------------------------------------------------------------------
+ # Helpers
+ # ------------------------------------------------------------------
+
+ @classmethod
+ def _spec(cls, name):
+ from pyspark.sql.mcp.tools.registry import all_tools
+
+ for spec in all_tools():
+ if spec.name == name:
+ return spec
+ raise KeyError(name)
+
+ @classmethod
+ def _call(cls, name, args=None):
+ spec = cls._spec(name)
+ return asyncio.run(spec.handler(args or {}, cls.holder))
+
+ # ------------------------------------------------------------------
+ # Tests
+ # ------------------------------------------------------------------
+
+ def test_get_session_info_reports_real_version(self):
+ out = self._call("get_session_info")
+ self.assertTrue(out["spark_version"], "spark_version should not be empty")
+ # Sanity-check semver-ish prefix.
+ self.assertRegex(out["spark_version"], r"^\d+\.\d+")
+ self.assertEqual(out["read_only"], True)
+ self.assertIn("current_catalog", out)
+ self.assertIn("current_database", out)
+
+ def test_list_catalogs_includes_default(self):
+ out = self._call("list_catalogs")
+ names = [item.get("name") for item in out["items"]]
+ self.assertIn("spark_catalog", names)
+
+ def test_list_databases_includes_default(self):
+ out = self._call("list_databases")
+ names = [item.get("name") for item in out["items"]]
+ self.assertIn("default", names)
+
+ def test_list_tables_finds_seeded_table(self):
+ out = self._call("list_tables")
+ names = [item.get("name") for item in out["items"]]
+ self.assertIn("mcp_test_orders", names)
+
+ def test_describe_table_returns_columns(self):
+ out = self._call("describe_table", {"name": "mcp_test_orders"})
+ self.assertEqual(out["table"]["name"], "mcp_test_orders")
+ col_names = [c["name"] for c in out["columns"]]
+ self.assertEqual(col_names, ["id", "amount"])
+
+ def test_execute_sql_returns_rows(self):
+ out = self._call(
+ "execute_sql",
+ {"query": "SELECT id, amount FROM mcp_test_orders ORDER BY id"},
+ )
+ self.assertEqual(out["row_count"], 3)
+ self.assertFalse(out["truncated"])
+ ids = [row["id"] for row in out["rows"]]
+ self.assertEqual(ids, [1, 2, 3])
+ # Schema metadata round-trips through Connect.
+ names = [field["name"] for field in out["schema"]]
+ self.assertEqual(names, ["id", "amount"])
+
+ def test_execute_sql_truncates_at_limit(self):
+ out = self._call(
+ "execute_sql",
+ {
+ "query": "SELECT id FROM mcp_test_orders ORDER BY id",
+ "limit": 2,
+ },
+ )
+ self.assertEqual(out["row_count"], 2)
+ self.assertTrue(out["truncated"])
+
+ def test_execute_sql_blocks_ddl_in_read_only(self):
+ from pyspark.errors import PySparkValueError
+
+ with self.assertRaises(PySparkValueError) as ctx:
+ self._call("execute_sql", {"query": "DROP TABLE mcp_test_orders"})
+ self.assertEqual(ctx.exception.getCondition(), "MCP_READ_ONLY_VIOLATION")
+
+ def test_explain_query_returns_plan_text(self):
+ out = self._call(
+ "explain_query",
+ {"query": "SELECT COUNT(*) FROM mcp_test_orders", "mode": "formatted"},
+ )
+ self.assertEqual(out["mode"], "formatted")
+ self.assertIn("Aggregate", out["plan"])
+
+ def test_analyze_query_includes_schema_and_plans(self):
+ out = self._call("analyze_query", {"query": "SELECT id FROM mcp_test_orders"})
+ self.assertEqual([f["name"] for f in out["schema"]], ["id"])
+ self.assertIn("extended_plan", out)
+ self.assertIn("formatted_plan", out)
+
+ def test_preview_table_returns_capped_rows(self):
+ out = self._call("preview_table", {"name": "mcp_test_orders", "limit": 2})
+ self.assertEqual(out["row_count"], 2)
+
+
+if __name__ == "__main__":
+ unittest.main()
Review Comment:
**Non-blocking (P2):** With a bare `unittest.main()`, a failing `setUpClass` exits with status 5 on Python 3.12-3.14, and `python/run-tests.py` excludes status 5 ('no test ran', SPARK-46801) from failures. On the pull-request CI's default Python 3.12, anything that breaks the Connect session or the seeded table therefore makes this suite pass silently, and no test-report XML is written. `from pyspark.testing import main` handles exactly this case and is what the other sql test modules use. The hardcoded `.remote('local[2]')` also ignores `SPARK_CONNECT_TESTING_REMOTE`, which the shared Connect test base honours.
**Verification:**
- **Regression:** With a deliberately failing class setup, the module exits with a failure status rather than the 'no tests ran' status on Python 3.12.
- **Behavior:** The integration suite passes against a remote Connect server given by SPARK_CONNECT_TESTING_REMOTE.
##########
python/pyspark/sql/mcp/safety.py:
##########
@@ -0,0 +1,53 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import re
+
+from pyspark.errors import PySparkValueError
+
+# Read-only allow-list. Comments and leading whitespace are stripped before
+# matching. We deliberately keep this conservative — anything not on the list
+# is rejected when read_only is on.
+_READ_ONLY_LEAD = re.compile(
+ r"^(select|with|show|describe|desc|explain|use|set|reset)\b",
+ re.IGNORECASE,
+)
+
+# Strip /* ... */ block comments and -- line comments before matching.
+_BLOCK_COMMENT = re.compile(r"/\*.*?\*/", re.DOTALL)
+_LINE_COMMENT = re.compile(r"--[^\n]*")
+
+
+def assert_read_only(query: str) -> None:
+ """Reject queries that mutate state when the server is in read-only mode.
+
+ The check is best-effort and string-based; it is a guardrail, not a
+ security boundary. Real isolation belongs at the Spark cluster (catalog
+ permissions, workspace policies, etc.).
+ """
+ stripped = _LINE_COMMENT.sub("", _BLOCK_COMMENT.sub("", query)).strip()
Review Comment:
**Nit (P3):** For every `/*` with no later `*/`, the lazy `/\*.*?\*/` scan runs to the end of the string before the next start position is tried, so the cost is quadratic in the query length. With this pattern, `'SELECT 1 ' + '/*a' * k` took 0.05 s, 0.20 s and 0.77 s at 6, 12 and 24 KB, against under 1 ms for closed comments. The check runs on the event loop before any timeout, so one such query stalls the whole server. Applying the pattern only up to the last `*/`, or using a `str.find` loop, gives the same result in linear time.
**Verification:**
- **Regression:** A query of about 100 KB made of unclosed '/*' sequences is classified in milliseconds.
- **Behavior:** On representative inputs the stripped text is the same as the current regex's output.
##########
python/pyspark/sql/mcp/tools/query.py:
##########
@@ -0,0 +1,363 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""SQL execution and plan tools."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Awaitable, Callable, Dict, List, Optional, TypeVar
+
+from pyspark.errors import PySparkRuntimeError, PySparkValueError
+from pyspark.sql.mcp.safety import assert_read_only
+from pyspark.sql.mcp.session import SessionHolder
+from pyspark.sql.mcp.tools.catalog import _to_dict
+from pyspark.sql.mcp.tools.registry import ToolSpec
+
+
+T = TypeVar("T")
+
+
+async def _with_timeout(holder: SessionHolder, func: Callable[[], T], *, label: str) -> T:
+ """Run a blocking Spark Connect call in a worker thread under the configured timeout."""
+ timeout = holder.config.query_timeout_seconds
+ coro: Awaitable[T] = asyncio.to_thread(func)
+ if timeout and timeout > 0:
+ try:
+ return await asyncio.wait_for(coro, timeout=timeout)
+ except asyncio.TimeoutError as exc:
+ raise PySparkRuntimeError(
+ errorClass="MCP_QUERY_TIMEOUT",
+ messageParameters={"label": label, "timeout": str(timeout)},
+ ) from exc
+ return await coro
+
+
+# ---------------------------------------------------------------------------
+# list_functions
+# ---------------------------------------------------------------------------
+
+
+async def _handle_list_functions(args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ spark = holder.get()
+ database = args.get("database")
+ pattern = args.get("pattern")
+
+ kwargs: Dict[str, Any] = {}
+ if database is not None:
+ kwargs["dbName"] = database
+ if pattern is not None:
+ kwargs["pattern"] = pattern
+ funcs = list(spark.catalog.listFunctions(**kwargs))
+
+ limit = int(args.get("limit", holder.config.default_page_size))
+ offset = int(args.get("offset", 0))
+ total = len(funcs)
+ page = funcs[offset : offset + limit]
+ return {
+ "items": [_to_dict(f) for f in page],
+ "offset": offset,
+ "limit": limit,
+ "returned": len(page),
+ "total": total,
+ "truncated": offset + limit < total,
+ }
+
+
+def list_functions_spec() -> ToolSpec:
+ return ToolSpec(
+ name="list_functions",
+ description=(
+ "List user-defined and built-in functions visible in a database. "
+ "Useful when authoring SQL — surfaces both UDFs and the built-in "
+ "function library."
+ ),
+ input_schema={
+ "type": "object",
+ "properties": {
+ "database": {"type": "string"},
+ "pattern": {"type": "string"},
+ "limit": {"type": "integer", "minimum": 1, "maximum": 1000, "default": 100},
+ "offset": {"type": "integer", "minimum": 0, "default": 0},
+ },
+ "additionalProperties": False,
+ },
+ handler=_handle_list_functions,
+ )
+
+
+# ---------------------------------------------------------------------------
+# Result rendering helpers
+# ---------------------------------------------------------------------------
+
+
+def _row_to_dict(row: Any) -> Dict[str, Any]:
+ if hasattr(row, "asDict"):
+ return row.asDict(recursive=True)
+ if hasattr(row, "_asdict"):
+ return dict(row._asdict())
+ return dict(row)
+
+
+def _render_markdown(rows: List[Dict[str, Any]], schema_fields: List[str]) -> str:
+ if not schema_fields:
+ return "_(no columns)_"
+ header = "| " + " | ".join(schema_fields) + " |"
+ sep = "| " + " | ".join("---" for _ in schema_fields) + " |"
+ body = []
+ for row in rows:
+ body.append("| " + " | ".join(_md_cell(row.get(c)) for c in schema_fields) + " |")
+ return "\n".join([header, sep, *body])
+
+
+def _md_cell(value: Any) -> str:
+ if value is None:
+ return ""
+ text = str(value).replace("|", "\\|").replace("\n", " ")
+ return text
+
+
+def _schema_fields(schema: Any) -> List[str]:
+ try:
+ return [field.name for field in schema.fields]
+ except AttributeError:
+ return []
+
+
+# ---------------------------------------------------------------------------
+# execute_sql
+# ---------------------------------------------------------------------------
+
+
+async def _handle_execute_sql(args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ spark = holder.get()
+ query: str = args["query"]
+ sql_args: Optional[Any] = args.get("args")
+ fmt: str = args.get("format", "json")
+ limit_arg = args.get("limit", holder.config.default_page_size)
+ offset = int(args.get("offset", 0))
+
+ if holder.config.read_only:
+ assert_read_only(query)
+
+ capped_limit = min(int(limit_arg), int(holder.config.max_rows))
+ df = _sql(spark, query, sql_args)
Review Comment:
**Non-blocking (P2):** `spark.sql` here is an ExecutePlan RPC that analyzes the statement and fully executes commands, and the `df.schema` on the next line is an AnalyzePlan RPC. Both run synchronously on the event loop before `_with_timeout` starts, as do `holder.get()`, every `spark.catalog` call and the session-info reads. An INSERT ... SELECT under `--no-read-only`, a slow SHOW or DESCRIBE, or `list_tables` against a slow metastore therefore runs with no limit, despite the documented 'wall-clock cap per tool call'. While it runs, the stdio server cannot answer anything else.
This extends an [existing discussion](https://github.com/apache/spark/pull/55648#issuecomment-4377110887).
See **Shared repair plan 1** in the review body.
##########
python/pyspark/sql/mcp/tools/query.py:
##########
@@ -0,0 +1,363 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""SQL execution and plan tools."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Awaitable, Callable, Dict, List, Optional, TypeVar
+
+from pyspark.errors import PySparkRuntimeError, PySparkValueError
+from pyspark.sql.mcp.safety import assert_read_only
+from pyspark.sql.mcp.session import SessionHolder
+from pyspark.sql.mcp.tools.catalog import _to_dict
+from pyspark.sql.mcp.tools.registry import ToolSpec
+
+
+T = TypeVar("T")
+
+
+async def _with_timeout(holder: SessionHolder, func: Callable[[], T], *, label: str) -> T:
+ """Run a blocking Spark Connect call in a worker thread under the configured timeout."""
+ timeout = holder.config.query_timeout_seconds
+ coro: Awaitable[T] = asyncio.to_thread(func)
+ if timeout and timeout > 0:
+ try:
+ return await asyncio.wait_for(coro, timeout=timeout)
+ except asyncio.TimeoutError as exc:
+ raise PySparkRuntimeError(
+ errorClass="MCP_QUERY_TIMEOUT",
+ messageParameters={"label": label, "timeout": str(timeout)},
+ ) from exc
+ return await coro
+
+
+# ---------------------------------------------------------------------------
+# list_functions
+# ---------------------------------------------------------------------------
+
+
+async def _handle_list_functions(args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ spark = holder.get()
+ database = args.get("database")
+ pattern = args.get("pattern")
+
+ kwargs: Dict[str, Any] = {}
+ if database is not None:
+ kwargs["dbName"] = database
+ if pattern is not None:
+ kwargs["pattern"] = pattern
+ funcs = list(spark.catalog.listFunctions(**kwargs))
+
+ limit = int(args.get("limit", holder.config.default_page_size))
+ offset = int(args.get("offset", 0))
+ total = len(funcs)
+ page = funcs[offset : offset + limit]
+ return {
+ "items": [_to_dict(f) for f in page],
+ "offset": offset,
+ "limit": limit,
+ "returned": len(page),
+ "total": total,
+ "truncated": offset + limit < total,
+ }
+
+
+def list_functions_spec() -> ToolSpec:
+ return ToolSpec(
+ name="list_functions",
+ description=(
+ "List user-defined and built-in functions visible in a database. "
+ "Useful when authoring SQL — surfaces both UDFs and the built-in "
+ "function library."
+ ),
+ input_schema={
+ "type": "object",
+ "properties": {
+ "database": {"type": "string"},
+ "pattern": {"type": "string"},
+ "limit": {"type": "integer", "minimum": 1, "maximum": 1000, "default": 100},
+ "offset": {"type": "integer", "minimum": 0, "default": 0},
+ },
+ "additionalProperties": False,
+ },
+ handler=_handle_list_functions,
+ )
+
+
+# ---------------------------------------------------------------------------
+# Result rendering helpers
+# ---------------------------------------------------------------------------
+
+
+def _row_to_dict(row: Any) -> Dict[str, Any]:
+ if hasattr(row, "asDict"):
+ return row.asDict(recursive=True)
+ if hasattr(row, "_asdict"):
+ return dict(row._asdict())
+ return dict(row)
+
+
+def _render_markdown(rows: List[Dict[str, Any]], schema_fields: List[str]) -> str:
+ if not schema_fields:
+ return "_(no columns)_"
+ header = "| " + " | ".join(schema_fields) + " |"
+ sep = "| " + " | ".join("---" for _ in schema_fields) + " |"
+ body = []
+ for row in rows:
+ body.append("| " + " | ".join(_md_cell(row.get(c)) for c in schema_fields) + " |")
+ return "\n".join([header, sep, *body])
+
+
+def _md_cell(value: Any) -> str:
+ if value is None:
+ return ""
+ text = str(value).replace("|", "\\|").replace("\n", " ")
+ return text
+
+
+def _schema_fields(schema: Any) -> List[str]:
+ try:
+ return [field.name for field in schema.fields]
+ except AttributeError:
+ return []
+
+
+# ---------------------------------------------------------------------------
+# execute_sql
+# ---------------------------------------------------------------------------
+
+
+async def _handle_execute_sql(args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ spark = holder.get()
+ query: str = args["query"]
+ sql_args: Optional[Any] = args.get("args")
+ fmt: str = args.get("format", "json")
+ limit_arg = args.get("limit", holder.config.default_page_size)
+ offset = int(args.get("offset", 0))
+
+ if holder.config.read_only:
+ assert_read_only(query)
+
+ capped_limit = min(int(limit_arg), int(holder.config.max_rows))
+ df = _sql(spark, query, sql_args)
+ schema_fields = _schema_fields(df.schema)
+
+ page_df = df.offset(offset).limit(capped_limit + 1) if offset else df.limit(capped_limit + 1)
+ collected = await _with_timeout(holder, page_df.collect, label="execute_sql")
+ rows = [_row_to_dict(r) for r in collected]
+ truncated = len(rows) > capped_limit
+ rows = rows[:capped_limit]
+
+ payload: Dict[str, Any] = {
+ "row_count": len(rows),
+ "truncated": truncated,
+ "limit": capped_limit,
+ "offset": offset,
+ "schema": [
+ {"name": field.name, "type": str(field.dataType), "nullable": field.nullable}
+ for field in df.schema.fields
+ ],
+ }
+ if fmt == "json":
+ payload["rows"] = rows
+ elif fmt == "markdown":
+ payload["markdown"] = _render_markdown(rows, schema_fields)
+ else:
+ raise PySparkValueError(
+ errorClass="MCP_UNSUPPORTED_FORMAT",
+ messageParameters={
+ "format": str(fmt),
+ "allowed": "json, markdown",
+ },
+ )
+ return payload
+
+
+def _sql(spark: Any, query: str, sql_args: Optional[Any]) -> Any:
+ if sql_args is None:
+ return spark.sql(query)
+ return spark.sql(query, args=sql_args)
+
+
+def execute_sql_spec() -> ToolSpec:
+ return ToolSpec(
+ name="execute_sql",
+ description=(
+ "Run a SQL query and return rows. Read-only by default — only "
+ "SELECT/WITH/SHOW/DESCRIBE/EXPLAIN/USE/SET are accepted unless "
Review Comment:
**Nit (P3):** The allow-list in safety.py also accepts `desc` and `reset`, so this 'only' list understates what read-only mode lets through, and RESET, like SET, changes the session configuration. Listing exactly what the allow-list accepts keeps the LLM-facing contract accurate.
**Verification:**
- **Inspection:** The description's list of accepted statement types equals the read-only allow-list.
##########
python/pyspark/sql/mcp/README.md:
##########
@@ -0,0 +1,186 @@
+# Spark MCP Server
+
+A [Model Context Protocol](https://modelcontextprotocol.io) server for Apache
+Spark, implemented as a thin client over [Spark
+Connect](https://spark.apache.org/docs/latest/spark-connect-overview.html).
+
+## Why
+
+LLM clients can already talk to MCP servers; Spark Connect already separates
+client from cluster. This module connects the two: a Spark cluster shows up
+to an LLM as a set of safe, paginated tools — `list_tables`, `describe_table`,
+`execute_sql`, `explain_query`, etc. — without giving the LLM a JVM driver.
+
+## Run
+
+The server speaks the `stdio` MCP transport (other transports may be added
+later) and connects to a Spark cluster via Spark Connect:
+
+```bash
+python -m pyspark.sql.mcp \
+ --connect-url "sc://localhost:15002"
+```
+
+Equivalent via environment variable:
+
+```bash
+export SPARK_REMOTE="sc://localhost:15002"
+python -m pyspark.sql.mcp
+```
+
+The server is **read-only by default**. Pass `--no-read-only` to allow DDL/DML
+(use with care; the read-only filter is a guardrail, not a security boundary).
+
+## Configuring an MCP client
+
+The server speaks `stdio` MCP transport, so a client launches it as a
+subprocess. The exact configuration step is client-specific; the moving
+parts are always:
+
+1. The command (`python -m pyspark.sql.mcp`).
+2. The Spark Connect URL, supplied via `--connect-url` or the
+ `SPARK_REMOTE` environment variable.
+3. Optionally, `--no-read-only` to permit DDL/DML.
+
+### Claude Code
+
+```bash
+claude mcp add spark \
+ -e SPARK_REMOTE=sc://localhost:15002 \
+ -- python -m pyspark.sql.mcp
+```
+
+Pass `--scope user` to register the server for all projects, or
+`--scope project` to write a shared `.mcp.json` into the current
+repository. Restart the Claude Code session and the LLM gains
+`mcp__spark__*` tools.
+
+### Claude Desktop
+
+Edit the application's `claude_desktop_config.json` (location is
+documented in the Claude Desktop docs) and add an entry under
+`mcpServers`:
+
+```json
+{
+ "mcpServers": {
+ "spark": {
+ "command": "python",
+ "args": ["-m", "pyspark.sql.mcp"],
+ "env": {
+ "SPARK_REMOTE": "sc://localhost:15002"
+ }
+ }
+ }
+}
+```
+
+Restart Claude Desktop.
+
+### Other MCP clients
+
+Any MCP-compliant client that can launch a stdio subprocess can use the
+same shape:
+
+```json
+{
+ "command": "python",
+ "args": ["-m", "pyspark.sql.mcp"],
+ "env": {"SPARK_REMOTE": "sc://localhost:15002"}
+}
+```
+
+### Environment variables
+
+| Variable | Meaning |
+| --- | --- |
+| `SPARK_REMOTE` | Spark Connect URL (`sc://host:port[/;params]`). |
+| `SPARK_MCP_READ_ONLY` | `true` (default) or `false`. |
+| `SPARK_MCP_USER_ID` | Optional Spark Connect session user id. |
+| `SPARK_MCP_MAX_ROWS` | Hard cap on rows returned per tool call (default `1000`). |
Review Comment:
**Nit (P3):** Only execute_sql, and preview_table through it, clamps with `min(limit, max_rows)`. list_catalogs, list_databases, list_tables and list_functions page with the caller's `limit`, so with `SPARK_MCP_MAX_ROWS=50` a listing called with `limit=1000` still returns up to 1000 items. Operators who set this variable to bound response size get no bound on those tools. Either apply the cap in every tool, or document which tools it covers.
**Verification:**
- **Behavior:** With max_rows set to 50, a listing requested with limit 1000 returns at most 50 items and reports that it was truncated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56698][PYTHON] Add Spark MCP (Model Context Protocol) server [spark]
Posted by "spark-test (via GitHub) via reviews" <re...@spark.apache.org>.
spark-test commented on code in PR #55648:
URL: https://github.com/apache/spark/pull/55648#discussion_r4509112118
##########
python/pyspark/sql/mcp/tools/catalog.py:
##########
@@ -0,0 +1,218 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Catalog browsing tools."""
+
+from __future__ import annotations
+
+from dataclasses import asdict, is_dataclass
+from typing import Any, Dict, List
+
+from pyspark.sql.mcp.session import SessionHolder
+from pyspark.sql.mcp.tools.registry import ToolSpec
+
+
+# ---------------------------------------------------------------------------
+# Result shaping
+# ---------------------------------------------------------------------------
+
+
+def _to_dict(obj: Any) -> Dict[str, Any]:
+ """Best-effort conversion of catalog dataclasses (CatalogMetadata, Database,
+ Table, Column, Function) into plain dicts."""
+ if is_dataclass(obj) and not isinstance(obj, type):
+ return asdict(obj)
+ if hasattr(obj, "_asdict"): # NamedTuple
+ return dict(obj._asdict())
+ if isinstance(obj, dict):
+ return obj
+ # Fallback: scrape public attributes.
+ return {
+ k: getattr(obj, k)
+ for k in dir(obj)
+ if not k.startswith("_") and not callable(getattr(obj, k))
+ }
+
+
+def _paginate(items: List[Any], limit: int, offset: int) -> Dict[str, Any]:
+ total = len(items)
+ end = offset + limit if limit is not None else total
+ page = items[offset:end]
+ return {
+ "items": [_to_dict(x) for x in page],
+ "offset": offset,
+ "limit": limit,
+ "returned": len(page),
+ "total": total,
+ "truncated": end < total,
+ }
+
+
+# ---------------------------------------------------------------------------
+# list_catalogs
+# ---------------------------------------------------------------------------
+
+
+async def _handle_list_catalogs(args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ spark = holder.get()
+ catalogs = list(spark.catalog.listCatalogs())
+ limit = int(args.get("limit", holder.config.default_page_size))
+ offset = int(args.get("offset", 0))
+ return _paginate(catalogs, limit, offset)
+
+
+def list_catalogs_spec() -> ToolSpec:
+ return ToolSpec(
+ name="list_catalogs",
+ description="List catalogs available to the current Spark session.",
+ input_schema={
+ "type": "object",
+ "properties": {
+ "limit": {"type": "integer", "minimum": 1, "maximum": 1000, "default": 100},
+ "offset": {"type": "integer", "minimum": 0, "default": 0},
+ },
+ "additionalProperties": False,
+ },
+ handler=_handle_list_catalogs,
+ )
+
+
+# ---------------------------------------------------------------------------
+# list_databases
+# ---------------------------------------------------------------------------
+
+
+async def _handle_list_databases(args: Dict[str, Any], holder: SessionHolder) -> Dict[str, Any]:
+ spark = holder.get()
+ catalog = args.get("catalog")
+ pattern = args.get("pattern")
+
+ if catalog:
+ spark.catalog.setCurrentCatalog(catalog)
+ if pattern is not None:
+ databases = list(spark.catalog.listDatabases(pattern))
+ else:
+ databases = list(spark.catalog.listDatabases())
+
+ limit = int(args.get("limit", holder.config.default_page_size))
+ offset = int(args.get("offset", 0))
+ return _paginate(databases, limit, offset)
+
+
+def list_databases_spec() -> ToolSpec:
+ return ToolSpec(
+ name="list_databases",
+ description=(
+ "List databases (a.k.a. schemas) in the current or named catalog. "
+ "Optional name pattern uses Spark's LIKE semantics."
Review Comment:
**Nit (P3):** `listDatabases(pattern)` runs SHOW NAMESPACES, which filters with `StringUtils.filterPattern`: only `*` (any characters) and `|` (alternatives) are special, and the rest is a case-insensitive regex. An LLM that follows 'LIKE semantics' writes `sales%`, gets no matches, and may conclude that the databases do not exist. list_tables and list_functions take the same kind of pattern with no description; stating the real syntax in all three fixes it.
**Verification:**
- **Inspection:** The pattern descriptions match Spark's catalog pattern semantics, including `*`, `|` and case-insensitivity.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56698][PYTHON] Add Spark MCP (Model Context Protocol) server [spark]
Posted by "viirya (via GitHub) via reviews" <re...@spark.apache.org>.
viirya commented on PR #55648:
URL: https://github.com/apache/spark/pull/55648#issuecomment-6731784621
> Any progress on the SPIP @viirya ?
I am still waiting for some approvals.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org