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.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.Required(),
|
||||
@@ -79,17 +79,30 @@ func (d *Deps) SparkSubmitHandler(ctx context.Context, req mcp.CallToolRequest)
|
||||
if err != nil {
|
||||
return errResult("spark_submit: " + err.Error()), nil
|
||||
}
|
||||
queue, err := req.RequireString("queue")
|
||||
|
||||
executorMemory, err := req.RequireString("executor_memory")
|
||||
if err != nil {
|
||||
return errResult("spark_submit: " + err.Error()), nil
|
||||
}
|
||||
executorMemory, err := req.RequireString("executor_memory")
|
||||
queue, err := req.RequireString("queue")
|
||||
if err != nil {
|
||||
return errResult("spark_submit: " + err.Error()), nil
|
||||
}
|
||||
|
||||
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
|
||||
// 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.
|
||||
|
||||
@@ -97,11 +97,16 @@ func TestSparkSubmit_StructuredCommand(t *testing.T) {
|
||||
|
||||
func TestSparkSubmit_MissingRequiredField(t *testing.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{
|
||||
"cluster_id": "cluster-echo",
|
||||
"deploy_mode": "cluster",
|
||||
"script_path": "/tmp/script.py",
|
||||
"script_path": mintedPath,
|
||||
"queue": "default",
|
||||
"executor_memory": "2G",
|
||||
"executor_cores": 1,
|
||||
@@ -125,12 +130,17 @@ func TestSparkSubmit_MissingRequiredField(t *testing.T) {
|
||||
|
||||
func TestSparkSubmit_BadSparkConfValue(t *testing.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{
|
||||
"cluster_id": "cluster-echo",
|
||||
"master": "yarn",
|
||||
"deploy_mode": "cluster",
|
||||
"script_path": "/tmp/script.py",
|
||||
"script_path": mintedPath,
|
||||
"queue": "default",
|
||||
"executor_memory": "2G",
|
||||
"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)
|
||||
}
|
||||
}
|
||||
|
||||
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