Compare commits
3
Commits
7920dc9597
...
67d1e9a6c5
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
67d1e9a6c5 | ||
|
|
44c9f415b7 | ||
|
|
c9e54a3f27 |
@@ -32,7 +32,7 @@ func NewSparkSubmitTool() mcp.Tool {
|
|||||||
),
|
),
|
||||||
mcp.WithString("script_path",
|
mcp.WithString("script_path",
|
||||||
mcp.Required(),
|
mcp.Required(),
|
||||||
mcp.Description("Absolute path to the script. Pass the path field returned by upload_file — do not construct your own path. The script is always the last argv element."),
|
mcp.Description("Absolute path to the script. Pass the `path` field returned by upload_file — do not construct your own path. The script is always the last argv element. Paths not minted by upload_file on this server are rejected."),
|
||||||
),
|
),
|
||||||
mcp.WithString("queue",
|
mcp.WithString("queue",
|
||||||
mcp.Required(),
|
mcp.Required(),
|
||||||
@@ -79,17 +79,30 @@ func (d *Deps) SparkSubmitHandler(ctx context.Context, req mcp.CallToolRequest)
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return errResult("spark_submit: " + err.Error()), nil
|
return errResult("spark_submit: " + err.Error()), nil
|
||||||
}
|
}
|
||||||
queue, err := req.RequireString("queue")
|
|
||||||
|
executorMemory, err := req.RequireString("executor_memory")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return errResult("spark_submit: " + err.Error()), nil
|
return errResult("spark_submit: " + err.Error()), nil
|
||||||
}
|
}
|
||||||
executorMemory, err := req.RequireString("executor_memory")
|
queue, err := req.RequireString("queue")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return errResult("spark_submit: " + err.Error()), nil
|
return errResult("spark_submit: " + err.Error()), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
args := req.GetArguments()
|
args := req.GetArguments()
|
||||||
|
|
||||||
|
// Reject paths that didn't come through upload_file on this server.
|
||||||
|
// Cluster-local paths and arbitrary host paths are unreachable from the MCP
|
||||||
|
// server's spark-submit, so we fail fast with a clear message rather than
|
||||||
|
// letting spark-submit produce a confusing FileNotFoundException later.
|
||||||
|
// Placed after all required-string parses so error messages surface in
|
||||||
|
// field order (cluster_id, master, ..., script_path, queue, ...).
|
||||||
|
if d.UploadStore != nil {
|
||||||
|
if _, err := d.UploadStore.Validate(scriptPath); err != nil {
|
||||||
|
return errResult(fmt.Sprintf("spark_submit: script_path %q is not from upload_file on this server; upload the file first via upload_file and pass back the path field", scriptPath)), nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// deploy_mode is required for forward compatibility but the builder does not
|
// deploy_mode is required for forward compatibility but the builder does not
|
||||||
// emit --deploy-mode, matching the Python reference's actual behavior. If the
|
// emit --deploy-mode, matching the Python reference's actual behavior. If the
|
||||||
// Go side needs --deploy-mode later, add it here and document the divergence.
|
// Go side needs --deploy-mode later, add it here and document the divergence.
|
||||||
|
|||||||
@@ -97,11 +97,16 @@ func TestSparkSubmit_StructuredCommand(t *testing.T) {
|
|||||||
|
|
||||||
func TestSparkSubmit_MissingRequiredField(t *testing.T) {
|
func TestSparkSubmit_MissingRequiredField(t *testing.T) {
|
||||||
deps, _ := testDepsWithDataDir(t)
|
deps, _ := testDepsWithDataDir(t)
|
||||||
|
store := deps.UploadStore
|
||||||
|
_, _, _, _, mintedPath, err := store.Save([]byte("# dummy\n"), "dummy.py")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("save upload: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
req := newToolRequest(SparkSubmitName, map[string]any{
|
req := newToolRequest(SparkSubmitName, map[string]any{
|
||||||
"cluster_id": "cluster-echo",
|
"cluster_id": "cluster-echo",
|
||||||
"deploy_mode": "cluster",
|
"deploy_mode": "cluster",
|
||||||
"script_path": "/tmp/script.py",
|
"script_path": mintedPath,
|
||||||
"queue": "default",
|
"queue": "default",
|
||||||
"executor_memory": "2G",
|
"executor_memory": "2G",
|
||||||
"executor_cores": 1,
|
"executor_cores": 1,
|
||||||
@@ -125,12 +130,17 @@ func TestSparkSubmit_MissingRequiredField(t *testing.T) {
|
|||||||
|
|
||||||
func TestSparkSubmit_BadSparkConfValue(t *testing.T) {
|
func TestSparkSubmit_BadSparkConfValue(t *testing.T) {
|
||||||
deps, _ := testDepsWithDataDir(t)
|
deps, _ := testDepsWithDataDir(t)
|
||||||
|
store := deps.UploadStore
|
||||||
|
_, _, _, _, mintedPath, err := store.Save([]byte("# dummy\n"), "dummy.py")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("save upload: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
req := newToolRequest(SparkSubmitName, map[string]any{
|
req := newToolRequest(SparkSubmitName, map[string]any{
|
||||||
"cluster_id": "cluster-echo",
|
"cluster_id": "cluster-echo",
|
||||||
"master": "yarn",
|
"master": "yarn",
|
||||||
"deploy_mode": "cluster",
|
"deploy_mode": "cluster",
|
||||||
"script_path": "/tmp/script.py",
|
"script_path": mintedPath,
|
||||||
"queue": "default",
|
"queue": "default",
|
||||||
"executor_memory": "2G",
|
"executor_memory": "2G",
|
||||||
"executor_cores": 1,
|
"executor_cores": 1,
|
||||||
@@ -152,3 +162,37 @@ func TestSparkSubmit_BadSparkConfValue(t *testing.T) {
|
|||||||
t.Errorf("error text=%q, want mention of spark_conf", text.Text)
|
t.Errorf("error text=%q, want mention of spark_conf", text.Text)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestSparkSubmit_RejectsNonMintedPath(t *testing.T) {
|
||||||
|
deps, _ := testDepsWithDataDir(t)
|
||||||
|
|
||||||
|
unminted := filepath.Join(t.TempDir(), "unminted.py")
|
||||||
|
if err := os.WriteFile(unminted, []byte("print('not from upload_file')\n"), 0o644); err != nil {
|
||||||
|
t.Fatalf("write unminted file: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
req := newToolRequest(SparkSubmitName, map[string]any{
|
||||||
|
"cluster_id": "cluster-echo",
|
||||||
|
"master": "yarn",
|
||||||
|
"deploy_mode": "cluster",
|
||||||
|
"script_path": unminted,
|
||||||
|
"queue": "default",
|
||||||
|
"executor_memory": "2G",
|
||||||
|
"executor_cores": 1,
|
||||||
|
"num_executors": 4,
|
||||||
|
})
|
||||||
|
res, err := deps.SparkSubmitHandler(context.Background(), req)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("handler error: %v", err)
|
||||||
|
}
|
||||||
|
if !res.IsError {
|
||||||
|
t.Fatalf("expected error result, got: %v", res.Content)
|
||||||
|
}
|
||||||
|
text, ok := mcp.AsTextContent(res.Content[0])
|
||||||
|
if !ok {
|
||||||
|
t.Fatalf("content is not text: %T", res.Content[0])
|
||||||
|
}
|
||||||
|
if !strings.Contains(text.Text, "not from upload_file") {
|
||||||
|
t.Errorf("error text=%q, want mention of not from upload_file", text.Text)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user