diff --git a/pkg/mcp/instructions.md b/pkg/mcp/instructions.md index 45af704c81..e46e27d7d0 100644 --- a/pkg/mcp/instructions.md +++ b/pkg/mcp/instructions.md @@ -34,6 +34,8 @@ This is essential because: - `deploy` tool: path = absolute path to Function directory (where func.yaml exists) - `build` tool: path = absolute path to Function directory (where func.yaml exists) - `config_*_list`, `config_*_add`, `config_*_remove` tools: path = absolute path to Function directory (where func.yaml exists) +- `run` tool: path = absolute path to Function directory (where func.yaml exists) +- `run_stop` tool: path = the SAME absolute path passed to `run` when starting it **IMPORTANT:** You must use absolute paths (e.g., `/Users/name/myproject/myfunc`), NOT relative paths (e.g., `.` or `myfunc`). The MCP server process runs in a different directory than your current working directory, so relative paths will not resolve correctly. @@ -58,6 +60,7 @@ This is essential because: - Before 'build' → Read `func://help/build` - Before 'list' → Read `func://help/list` - Before 'delete' → Read `func://help/delete` +- Before 'run' → Read `func://help/run` The help text provides authoritative parameter information and usage context. @@ -140,6 +143,28 @@ A first-time deploy can be detected by checking the func.yaml for a value in the - Exactly ONE of 'path' or 'name' must be provided, not both - Deleting does not affect local files (source). Only cluster resources. +### run + +- **FIRST:** Read `func://help/run` for authoritative usage information +- **REQUIRED parameters:** + - `path` (absolute path to the directory containing the Function to run; there is no CWD-based default — the MCP server's own working directory is unrelated to yours) +- **OPTIONAL parameters:** + - `registry` (container registry; only needed if the Function's image must be named/built) + - `build` (force a rebuild before running; omit to let func build automatically only when out of date) + - `port` (host port to bind; omit to use 8080 or the first available port) +- Builds the Function locally if needed, starts it, and returns once it is up: `pid` (process ID) and `url` +- The Function keeps running in the background after the tool call returns +- Only ONE run may be active per Function path at a time — calling `run` again for a path that is already running fails with a clear error; call `run_stop` first +- ALWAYS call `run_stop` with the matching `path` when finished, to free the port and clean up +- Not a cluster operation, so unaffected by read-only mode + +### run_stop + +- Stops a Function previously started with `run` +- **REQUIRED:** `path` (absolute) must match the path used to start it with `run` +- Idempotent: calling `run_stop` for a path with no active run (already stopped, or never started) succeeds with an informational message rather than erroring +- Not a cluster operation, so unaffected by read-only mode + ### config_envs_list, config_envs_add, config_envs_remove - **BEFORE calling add/remove:** Consider reading `func://help/config/envs` for authoritative usage diff --git a/pkg/mcp/instructions_warning.md b/pkg/mcp/instructions_warning.md index 98697ddeec..9e3fded82f 100644 --- a/pkg/mcp/instructions_warning.md +++ b/pkg/mcp/instructions_warning.md @@ -11,6 +11,7 @@ The Functions MCP server is currently running in **read-only mode**. - Build Functions - Configure Functions (envs, labels, volumes) - Inspect Functions +- Run/stop Functions locally (not cluster operations, so unaffected by read-only mode) **Disabled operations:** - Deploy to cluster diff --git a/pkg/mcp/mcp.go b/pkg/mcp/mcp.go index 02ea0f5d44..3cecfc8a54 100644 --- a/pkg/mcp/mcp.go +++ b/pkg/mcp/mcp.go @@ -26,8 +26,10 @@ type Server struct { prefix string // Command prefix ("func" or "kn func") readonly atomic.Bool // disables deploy and delete when true executor executor - transport mcp.Transport // Transport to use (defaults to StdioTransport) - impl *mcp.Server // implements the protocol + transport mcp.Transport // Transport to use (defaults to StdioTransport) + impl *mcp.Server // implements the protocol + starter processStarter // starts long-lived "func run" subprocesses + runs *runRegistry // tracks active local runs, keyed by function path } type executor interface { @@ -61,6 +63,14 @@ func WithExecutor(executor executor) Option { } } +// WithProcessStarter sets a custom process starter for the "run" tool; used +// in tests. +func WithProcessStarter(starter processStarter) Option { + return func(s *Server) { + s.starter = starter + } +} + // WithTransport sets a custom transport for the server; used in tests. func WithTransport(transport mcp.Transport) Option { return func(s *Server) { @@ -83,6 +93,8 @@ func New(options ...Option) *Server { OnInit: func(_ context.Context) {}, } s.executor = defaultExecutor{s} + s.starter = defaultProcessStarter{s} + s.runs = newRunRegistry() for _, o := range options { o(s) } @@ -109,6 +121,8 @@ func New(options ...Option) *Server { mcp.AddTool(i, deployTool, s.deployHandler) mcp.AddTool(i, listTool, s.listHandler) mcp.AddTool(i, deleteTool, s.deleteHandler) + mcp.AddTool(i, runTool, s.runHandler) + mcp.AddTool(i, runStopTool, s.runStopHandler) mcp.AddTool(i, configVolumesListTool, s.configVolumesListHandler) mcp.AddTool(i, configVolumesAddTool, s.configVolumesAddHandler) mcp.AddTool(i, configVolumesRemoveTool, s.configVolumesRemoveHandler) @@ -160,8 +174,16 @@ func New(options ...Option) *Server { // Start the MCP server using the configured transport. // The server's readonly mode is determined at construction time via // WithReadonly; it cannot be changed after the server is created. +// +// When Run returns, on normal shutdown (client disconnect, or the caller +// canceling ctx in response to SIGINT/SIGTERM), any Function runs left +// active by the "run" tool are stopped so no subprocess (and the port it +// holds) is left behind. A hard kill of the server process itself (e.g. a +// second SIGKILL) bypasses this; the OS reaps the orphaned children instead. func (s *Server) Start(ctx context.Context) error { - return s.impl.Run(ctx, s.transport) + err := s.impl.Run(ctx, s.transport) + s.runs.stopAll() + return err } // For now the executor is a simple run of the command "func" or "kn func" diff --git a/pkg/mcp/mock/process_starter.go b/pkg/mcp/mock/process_starter.go new file mode 100644 index 0000000000..22d33151c7 --- /dev/null +++ b/pkg/mcp/mock/process_starter.go @@ -0,0 +1,31 @@ +package mock + +import ( + "context" +) + +// ProcessStarter is a mock implementation of the process-starter interface +// used by the "run" tool. It avoids spawning real subprocesses in tests. +// It implements the same interface as mcp.processStarter through structural +// typing. +type ProcessStarter struct { + StartInvoked bool + StartFn func(ctx context.Context, subcommand string, args ...string) (pid int, host, port string, stop func() error, err error) +} + +// NewProcessStarter creates a new mock process starter. +func NewProcessStarter() *ProcessStarter { + return &ProcessStarter{} +} + +// Start implements the processStarter interface, recording invocation +// details and delegating to StartFn if provided. +func (m *ProcessStarter) Start(ctx context.Context, subcommand string, args ...string) (pid int, host, port string, stop func() error, err error) { + m.StartInvoked = true + + if m.StartFn != nil { + return m.StartFn(ctx, subcommand, args...) + } + + return 1234, "127.0.0.1", "8080", func() error { return nil }, nil +} diff --git a/pkg/mcp/process.go b/pkg/mcp/process.go new file mode 100644 index 0000000000..98da0008bd --- /dev/null +++ b/pkg/mcp/process.go @@ -0,0 +1,312 @@ +package mcp + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "io" + "os/exec" + "path/filepath" + "strings" + "sync" + "syscall" + "time" +) + +// runReadyTimeout bounds how long "run" waits for the started function to +// report readiness (build + container/host startup can be slow on a cold +// cache). A var (rather than a const) so tests can shrink it. +var runReadyTimeout = 3 * time.Minute + +// tailBufferSize is the number of recent output lines retained for +// inclusion in diagnostic error messages. +const tailBufferSize = 200 + +// processStarter starts a long-lived background subcommand (e.g. "func +// run") and waits for it to signal readiness: a line of JSON containing +// non-empty "host" and "port" fields on stdout (produced by passing +// --json). It is abstracted so tests can inject a fake implementation +// instead of spawning real subprocesses. +type processStarter interface { + // Start begins running subcommand with args in the background. It + // blocks until the process reports readiness, exits, errors, or ctx is + // done, whichever happens first. + // + // On success it returns the process ID, host, and port, along with a + // stop function which gracefully terminates the process (SIGTERM only, + // no forced kill) and blocks until it has fully exited. stop is + // idempotent and safe to call even if the process has already exited on + // its own. + // + // On error, any process that was started is terminated before + // returning; no process is left running. + Start(ctx context.Context, subcommand string, args ...string) (pid int, host, port string, stop func() error, err error) +} + +// defaultProcessStarter starts subprocesses using the server's configured +// command prefix (e.g. "func" or "kn func"). +type defaultProcessStarter struct { + s *Server +} + +func (d defaultProcessStarter) Start(ctx context.Context, subcommand string, args ...string) (pid int, host, port string, stop func() error, err error) { + cmdParts := buildArgs(d.s.prefix, subcommand, args) + + // The subprocess must outlive this call (and the request context that + // bounds it), so it is given its own independent, cancelable context. + // The incoming ctx is used only to bound how long we wait below for + // readiness. + procCtx, cancel := context.WithCancel(context.Background()) + + cmd := exec.CommandContext(procCtx, cmdParts[0], cmdParts[1:]...) + // Send SIGTERM (not the default SIGKILL) on cancellation so the child's + // own signal handler can run its cleanup (job.Stop(), container + // teardown, etc.), and wait for it to exit on its own. We deliberately + // do not escalate to SIGKILL: a process that never exits after SIGTERM + // is a bug in the runner to fix there, not something for the MCP server + // to paper over by force-killing it (which would skip that cleanup). + cmd.Cancel = func() error { + if cmd.Process == nil { + return nil + } + return cmd.Process.Signal(syscall.SIGTERM) + } + + stdout, err := cmd.StdoutPipe() + if err != nil { + cancel() + return 0, "", "", nil, fmt.Errorf("unable to create stdout pipe: %w", err) + } + tail := newTailBuffer(tailBufferSize) + cmd.Stderr = tail + + if err = cmd.Start(); err != nil { + cancel() + return 0, "", "", nil, fmt.Errorf("unable to start %q: %w", strings.Join(cmdParts, " "), err) + } + + exited := make(chan struct{}) + var waitErr error + go func() { + waitErr = cmd.Wait() + close(exited) + }() + + ready := make(chan readyResult, 1) + go scanForReady(stdout, tail, ready) + + var stopOnce sync.Once + stop = func() error { + stopOnce.Do(func() { + select { + case <-exited: + return // already exited; nothing to signal + default: + } + cancel() + <-exited + }) + return nil + } + + select { + case r := <-ready: + return cmd.Process.Pid, r.host, r.port, stop, nil + case <-exited: + _ = stop() + return 0, "", "", nil, fmt.Errorf("process exited before becoming ready (%v)\noutput:\n%s", waitErr, tail.String()) + case <-ctx.Done(): + _ = stop() + return 0, "", "", nil, fmt.Errorf("timed out waiting for function to become ready\noutput so far:\n%s", tail.String()) + } +} + +// readyResult carries the host/port parsed from a subprocess's --json +// readiness line. +type readyResult struct { + host, port string +} + +// runReadyLine is the shape of the single line of JSON that "func run +// --json" prints once the function is up and healthy. +type runReadyLine struct { + Host string `json:"host"` + Port string `json:"port"` +} + +// scanForReady drains r line-by-line for the lifetime of the process +// (required to avoid the child blocking on a full stdout pipe once it +// starts streaming logs), sending exactly once on ready as soon as a valid +// readiness line is found. Every line is also recorded in tail for +// diagnostics. It returns once r reaches EOF (i.e. the process closed +// stdout, generally because it exited). +func scanForReady(r io.ReadCloser, tail *tailBuffer, ready chan<- readyResult) { + defer r.Close() + sent := false + scanner := bufio.NewScanner(r) + scanner.Buffer(make([]byte, 64*1024), 1<<20) + for scanner.Scan() { + line := scanner.Text() + tail.WriteLine("stdout: " + line) + if !sent { + if host, port, ok := parseReadyLine(line); ok { + sent = true + ready <- readyResult{host: host, port: port} + } + } + } +} + +// parseReadyLine attempts to parse line as the --json readiness output. +func parseReadyLine(line string) (host, port string, ok bool) { + var out runReadyLine + if err := json.Unmarshal([]byte(line), &out); err != nil { + return "", "", false + } + if out.Host == "" || out.Port == "" { + return "", "", false + } + return out.Host, out.Port, true +} + +// tailBuffer is a bounded, concurrency-safe buffer of the most recent output +// lines from a subprocess, kept for inclusion in error messages. It +// implements io.Writer so it can be used directly as a Cmd's Stderr. +type tailBuffer struct { + mu sync.Mutex + lines []string + max int +} + +func newTailBuffer(max int) *tailBuffer { + return &tailBuffer{max: max} +} + +func (t *tailBuffer) WriteLine(line string) { + t.mu.Lock() + defer t.mu.Unlock() + t.lines = append(t.lines, line) + if len(t.lines) > t.max { + t.lines = t.lines[len(t.lines)-t.max:] + } +} + +func (t *tailBuffer) Write(p []byte) (int, error) { + if len(p) == 0 { + return 0, nil + } + for _, line := range strings.Split(strings.TrimRight(string(p), "\n"), "\n") { + t.WriteLine("stderr: " + line) + } + return len(p), nil +} + +func (t *tailBuffer) String() string { + t.mu.Lock() + defer t.mu.Unlock() + return strings.Join(t.lines, "\n") +} + +// runEntry tracks a single "run" invocation. A freshly reserved entry has a +// nil stop (and zero pid) until activate fills it in once the subprocess has +// actually started and become ready; get treats such a pending entry as not +// yet active. +type runEntry struct { + pid int + stop func() error +} + +// runRegistry tracks active local function runs, keyed by the resolved +// absolute path of the Function being run. It is the single source of +// truth used by both the "run" and "run_stop" tools. +type runRegistry struct { + mu sync.Mutex + byPath map[string]*runEntry +} + +func newRunRegistry() *runRegistry { + return &runRegistry{byPath: map[string]*runEntry{}} +} + +// reserve atomically claims path for a new run before the (slow) subprocess +// Start call is made, so that two concurrent "run" calls for the same path +// cannot both spawn a process: the second is rejected here, before anything +// is started. The caller must follow a successful reserve with either +// activate (Start succeeded) or release (Start failed). +func (r *runRegistry) reserve(path string) error { + r.mu.Lock() + defer r.mu.Unlock() + if existing, ok := r.byPath[path]; ok { + if existing.stop == nil { + return fmt.Errorf("a function is already starting at %q; call run_stop first", path) + } + return fmt.Errorf("a function is already running at %q (pid %d); call run_stop first", path, existing.pid) + } + r.byPath[path] = &runEntry{} + return nil +} + +// activate fills in the pid/stop for a path previously reserve'd, marking it +// as an active run. +func (r *runRegistry) activate(path string, pid int, stop func() error) { + r.mu.Lock() + defer r.mu.Unlock() + r.byPath[path] = &runEntry{pid: pid, stop: stop} +} + +// release removes a reservation for path, used when Start fails after a +// successful reserve. +func (r *runRegistry) release(path string) { + r.remove(path) +} + +// get returns the active run entry for path, if any. A path that is +// reserved but not yet activated (still starting) is treated as not active. +func (r *runRegistry) get(path string) (*runEntry, bool) { + r.mu.Lock() + defer r.mu.Unlock() + e, ok := r.byPath[path] + if !ok || e.stop == nil { + return nil, false + } + return e, true +} + +func (r *runRegistry) remove(path string) { + r.mu.Lock() + defer r.mu.Unlock() + delete(r.byPath, path) +} + +// stopAll stops every currently active run and clears the registry. It is +// called on normal server shutdown (client disconnect, or SIGINT/SIGTERM +// canceling the server's context) so that no "func run" subprocess is left +// behind holding a port. It does nothing useful on a hard kill of the +// server itself (e.g. SIGKILL), where OS process reaping applies instead. +func (r *runRegistry) stopAll() { + r.mu.Lock() + entries := make([]*runEntry, 0, len(r.byPath)) + for _, e := range r.byPath { + entries = append(entries, e) + } + r.byPath = map[string]*runEntry{} + r.mu.Unlock() + + for _, e := range entries { + if e.stop != nil { + _ = e.stop() + } + } +} + +// resolveRunPath validates the required path input for the run/run_stop +// tools. The MCP server process's working directory is unrelated to the +// caller's, so path must be an absolute path supplied explicitly; there is +// no CWD-based default. +func resolveRunPath(path string) (string, error) { + if !filepath.IsAbs(path) { + return "", fmt.Errorf("path must be an absolute path, got %q", path) + } + return filepath.Clean(path), nil +} diff --git a/pkg/mcp/process_test.go b/pkg/mcp/process_test.go new file mode 100644 index 0000000000..02593f218c --- /dev/null +++ b/pkg/mcp/process_test.go @@ -0,0 +1,226 @@ +package mcp + +import ( + "context" + "fmt" + "os" + "path/filepath" + "runtime" + "strings" + "testing" + "time" +) + +// TestRunRegistry_ReserveActivateRemove verifies the basic lifecycle of a +// runRegistry entry, including rejection of a duplicate reserve for the same +// path both before and after activation. +func TestRunRegistry_ReserveActivateRemove(t *testing.T) { + r := newRunRegistry() + stopped := false + stop := func() error { stopped = true; return nil } + + if _, ok := r.get("/a/b"); ok { + t.Fatal("expected no entry before reserve") + } + + if err := r.reserve("/a/b"); err != nil { + t.Fatalf("unexpected error reserving: %v", err) + } + + // A pending (reserved but not yet activated) entry is not "active". + if _, ok := r.get("/a/b"); ok { + t.Fatal("expected no active entry while still pending") + } + + // A second reserve while pending must be rejected. + if err := r.reserve("/a/b"); err == nil { + t.Fatal("expected error reserving an already-pending path") + } + + r.activate("/a/b", 111, stop) + + entry, ok := r.get("/a/b") + if !ok { + t.Fatal("expected entry after activate") + } + if entry.pid != 111 { + t.Fatalf("expected pid 111, got %d", entry.pid) + } + + // A reserve for an already-active path must be rejected. + if err := r.reserve("/a/b"); err == nil { + t.Fatal("expected error reserving an already-active path") + } + + r.remove("/a/b") + if _, ok := r.get("/a/b"); ok { + t.Fatal("expected no entry after remove") + } + if err := entry.stop(); err != nil { + t.Fatalf("unexpected error calling stop: %v", err) + } + if !stopped { + t.Fatal("expected stop to have been invoked") + } +} + +// TestRunRegistry_StopAll verifies that stopAll stops every active run and +// clears the registry. +func TestRunRegistry_StopAll(t *testing.T) { + r := newRunRegistry() + var stoppedA, stoppedB bool + + if err := r.reserve("/a"); err != nil { + t.Fatal(err) + } + r.activate("/a", 1, func() error { stoppedA = true; return nil }) + + if err := r.reserve("/b"); err != nil { + t.Fatal(err) + } + r.activate("/b", 2, func() error { stoppedB = true; return nil }) + + r.stopAll() + + if !stoppedA || !stoppedB { + t.Fatalf("expected both runs to be stopped, got a=%v b=%v", stoppedA, stoppedB) + } + if _, ok := r.get("/a"); ok { + t.Fatal("expected registry to be cleared after stopAll") + } + if _, ok := r.get("/b"); ok { + t.Fatal("expected registry to be cleared after stopAll") + } +} + +// TestResolveRunPath verifies that a required absolute path is accepted +// as-is (cleaned), and that a relative path is rejected rather than +// resolved against the server process's working directory. +func TestResolveRunPath(t *testing.T) { + wd, err := os.Getwd() + if err != nil { + t.Fatal(err) + } + abs := filepath.Join(wd, "myfunc") + + got, err := resolveRunPath(abs) + if err != nil { + t.Fatal(err) + } + if got != abs { + t.Fatalf("expected %q, got %q", abs, got) + } + + if _, err = resolveRunPath(""); err == nil { + t.Fatal("expected error for empty path") + } + + if _, err = resolveRunPath("."); err == nil { + t.Fatal("expected error for relative path") + } + + if _, err = resolveRunPath("myfunc"); err == nil { + t.Fatal("expected error for relative path") + } +} + +// writeTestScript writes an executable shell script to a temp file that: +// - traps SIGTERM to exit cleanly (mimicking func run's graceful shutdown) +// - optionally sleeps before printing anything (to simulate slow startup) +// - prints the given stdout line(s) +// - then idles until signaled or killed +func writeTestScript(t *testing.T, preSleep time.Duration, stdoutLines ...string) string { + t.Helper() + if runtime.GOOS == "windows" { + t.Skip("test script requires a POSIX shell") + } + + dir := t.TempDir() + path := filepath.Join(dir, "fake-func.sh") + + var b strings.Builder + b.WriteString("#!/bin/sh\n") + b.WriteString("trap 'exit 0' TERM\n") + if preSleep > 0 { + fmt.Fprintf(&b, "sleep %f\n", preSleep.Seconds()) + } + for _, line := range stdoutLines { + fmt.Fprintf(&b, "echo '%s'\n", line) + } + b.WriteString("while true; do sleep 1; done\n") + + if err := os.WriteFile(path, []byte(b.String()), 0o755); err != nil { + t.Fatal(err) + } + return path +} + +func newTestStarter(t *testing.T, script string) defaultProcessStarter { + t.Helper() + return defaultProcessStarter{s: &Server{prefix: script}} +} + +// TestDefaultProcessStarter_Ready verifies that Start waits for the +// readiness line, then returns pid/host/port, and that the returned stop +// function terminates the process via SIGTERM. +func TestDefaultProcessStarter_Ready(t *testing.T) { + script := writeTestScript(t, 0, `{"host":"127.0.0.1","port":"9999"}`) + starter := newTestStarter(t, script) + + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + + pid, host, port, stop, err := starter.Start(ctx, "run") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if pid <= 0 { + t.Fatalf("expected positive pid, got %d", pid) + } + if host != "127.0.0.1" || port != "9999" { + t.Fatalf("expected 127.0.0.1:9999, got %s:%s", host, port) + } + + if err = stop(); err != nil { + t.Fatalf("unexpected error from stop: %v", err) + } + // stop must be idempotent. + if err = stop(); err != nil { + t.Fatalf("unexpected error from second stop call: %v", err) + } +} + +// TestDefaultProcessStarter_ExitsBeforeReady verifies that a process which +// exits without ever emitting a valid readiness line surfaces a clear error. +func TestDefaultProcessStarter_ExitsBeforeReady(t *testing.T) { + script := writeTestScript(t, 0) + // Override the idle loop with an immediate exit by writing our own + // script body instead of relying on writeTestScript's infinite loop. + if err := os.WriteFile(script, []byte("#!/bin/sh\necho 'not json'\nexit 1\n"), 0o755); err != nil { + t.Fatal(err) + } + starter := newTestStarter(t, script) + + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + + _, _, _, _, err := starter.Start(ctx, "run") + if err == nil { + t.Fatal("expected an error when the process exits before becoming ready") + } +} + +// TestDefaultProcessStarter_Timeout verifies that Start gives up and returns +// an error if the process never becomes ready within the given context. +func TestDefaultProcessStarter_Timeout(t *testing.T) { + script := writeTestScript(t, 3*time.Second, `{"host":"127.0.0.1","port":"9999"}`) + starter := newTestStarter(t, script) + + ctx, cancel := context.WithTimeout(t.Context(), 300*time.Millisecond) + defer cancel() + + _, _, _, _, err := starter.Start(ctx, "run") + if err == nil { + t.Fatal("expected a timeout error") + } +} diff --git a/pkg/mcp/tools_run.go b/pkg/mcp/tools_run.go new file mode 100644 index 0000000000..be56c86121 --- /dev/null +++ b/pkg/mcp/tools_run.go @@ -0,0 +1,83 @@ +package mcp + +import ( + "context" + "fmt" + + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +var runTool = &mcp.Tool{ + Name: "run", + Title: "Run Function Locally", + Description: "Run a Function locally (building it first if needed) and return its process ID and URL.", + Annotations: &mcp.ToolAnnotations{ + Title: "Run Function Locally", + ReadOnlyHint: false, + DestructiveHint: ptr(false), + IdempotentHint: false, // calling run again for a path already running is rejected, not idempotent + }, +} + +func (s *Server) runHandler(ctx context.Context, r *mcp.CallToolRequest, input RunInput) (result *mcp.CallToolResult, output RunOutput, err error) { + path, err := resolveRunPath(input.Path) + if err != nil { + err = fmt.Errorf("unable to resolve function path: %w", err) + return + } + + // reserve claims path before the subprocess is started, so that two + // concurrent "run" calls for the same path cannot both spawn a + // process; the loser is rejected here, before any process is started. + if err = s.runs.reserve(path); err != nil { + return + } + + readyCtx, cancel := context.WithTimeout(ctx, runReadyTimeout) + defer cancel() + + pid, host, port, stop, err := s.starter.Start(readyCtx, "run", input.Args(path)...) + if err != nil { + s.runs.release(path) + err = fmt.Errorf("unable to run function: %w", err) + return + } + + s.runs.activate(path, pid, stop) + + output = RunOutput{ + Pid: pid, + URL: fmt.Sprintf("http://%s:%s", host, port), + } + return +} + +// RunInput defines the input parameters for the run tool. +type RunInput struct { + Path string `json:"path" jsonschema:"required,Absolute path to the function project directory"` + Registry *string `json:"registry,omitempty" jsonschema:"Container registry for the function image"` + Build *bool `json:"build,omitempty" jsonschema:"Force a rebuild before running (default false; a build still happens automatically if the image is missing or out of date)"` + Port *int `json:"port,omitempty" jsonschema:"Host port to bind (default: 8080, or the first available port)"` +} + +// Args builds the "func run" argument list for the resolved, absolute path. +func (i RunInput) Args(path string) []string { + args := []string{"--path", path, "--json"} + + args = appendStringFlag(args, "--registry", i.Registry) + + if i.Build != nil && *i.Build { + args = append(args, "--build=true") + } + if i.Port != nil { + args = append(args, "--address", fmt.Sprintf("127.0.0.1:%d", *i.Port)) + } + + return args +} + +// RunOutput defines the structured output returned by the run tool. +type RunOutput struct { + Pid int `json:"pid" jsonschema:"Process ID of the running function"` + URL string `json:"url" jsonschema:"URL on which the function is listening"` +} diff --git a/pkg/mcp/tools_run_stop.go b/pkg/mcp/tools_run_stop.go new file mode 100644 index 0000000000..7b3610cd31 --- /dev/null +++ b/pkg/mcp/tools_run_stop.go @@ -0,0 +1,59 @@ +package mcp + +import ( + "context" + "fmt" + + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +var runStopTool = &mcp.Tool{ + Name: "run_stop", + Title: "Stop Local Function Run", + Description: "Stop a Function previously started with the run tool.", + Annotations: &mcp.ToolAnnotations{ + Title: "Stop Local Function Run", + ReadOnlyHint: false, + DestructiveHint: ptr(true), + IdempotentHint: true, // stopping an already-stopped run at the same path succeeds either way + }, +} + +func (s *Server) runStopHandler(ctx context.Context, r *mcp.CallToolRequest, input RunStopInput) (result *mcp.CallToolResult, output RunStopOutput, err error) { + path, err := resolveRunPath(input.Path) + if err != nil { + err = fmt.Errorf("unable to resolve function path: %w", err) + return + } + + entry, ok := s.runs.get(path) + if !ok { + // Idempotent: stopping a path with no active run (already stopped, + // or never started) is not an error. + output = RunStopOutput{ + Message: fmt.Sprintf("no active run found at %q; already stopped", path), + } + return + } + + if err = entry.stop(); err != nil { + err = fmt.Errorf("unable to stop function at %q: %w", path, err) + return + } + s.runs.remove(path) + + output = RunStopOutput{ + Message: fmt.Sprintf("stopped function at %q (pid %d)", path, entry.pid), + } + return +} + +// RunStopInput defines the input parameters for the run_stop tool. +type RunStopInput struct { + Path string `json:"path" jsonschema:"required,Absolute path to the function project directory; must match the path used with the run tool"` +} + +// RunStopOutput defines the structured output returned by the run_stop tool. +type RunStopOutput struct { + Message string `json:"message" jsonschema:"Confirmation message"` +} diff --git a/pkg/mcp/tools_run_stop_test.go b/pkg/mcp/tools_run_stop_test.go new file mode 100644 index 0000000000..bf35b14d9a --- /dev/null +++ b/pkg/mcp/tools_run_stop_test.go @@ -0,0 +1,108 @@ +package mcp + +import ( + "context" + "testing" + + "github.com/modelcontextprotocol/go-sdk/mcp" + "knative.dev/func/pkg/mcp/mock" +) + +// TestTool_RunStop_AllowedInReadonly ensures the run_stop tool is NOT gated +// by readonly mode: it is a local-only operation (no cluster mutation). +func TestTool_RunStop_AllowedInReadonly(t *testing.T) { + client, _, err := newTestPairWithReadonly(t, true) + if err != nil { + t.Fatal(err) + } + + result, err := client.CallTool(t.Context(), &mcp.CallToolParams{ + Name: "run_stop", + Arguments: map[string]any{"path": testAbsPath("my-func")}, + }) + if err != nil { + t.Fatal(err) + } + if result.IsError { + t.Fatalf("expected run_stop to be allowed in readonly mode, got error: %v", result) + } +} + +// TestTool_RunStop_NotRunning ensures run_stop is idempotent: stopping a +// path with no active run succeeds with an informational message rather +// than erroring. +func TestTool_RunStop_NotRunning(t *testing.T) { + client, server, err := newTestPair(t) + if err != nil { + t.Fatal(err) + } + server.readonly.Store(false) + + result, err := client.CallTool(t.Context(), &mcp.CallToolParams{ + Name: "run_stop", + Arguments: map[string]any{"path": testAbsPath("never-ran")}, + }) + if err != nil { + t.Fatal(err) + } + if result.IsError { + t.Fatalf("expected run_stop for a path with no active run to succeed idempotently, got error: %v", result) + } +} + +// TestTool_RunStop_Success ensures run_stop invokes the stop function +// returned by a prior run, removes the registry entry, and that a +// subsequent run_stop for the same path then fails clearly. +func TestTool_RunStop_Success(t *testing.T) { + stopped := false + starter := mock.NewProcessStarter() + starter.StartFn = func(ctx context.Context, subcommand string, args ...string) (int, string, string, func() error, error) { + return 555, "127.0.0.1", "8080", func() error { stopped = true; return nil }, nil + } + + client, server, err := newTestPair(t, WithProcessStarter(starter)) + if err != nil { + t.Fatal(err) + } + server.readonly.Store(false) + + path := testAbsPath("stop-me") + + result, err := client.CallTool(t.Context(), &mcp.CallToolParams{ + Name: "run", + Arguments: map[string]any{"path": path}, + }) + if err != nil { + t.Fatal(err) + } + if result.IsError { + t.Fatalf("unexpected error starting run: %v", result) + } + + result, err = client.CallTool(t.Context(), &mcp.CallToolParams{ + Name: "run_stop", + Arguments: map[string]any{"path": path}, + }) + if err != nil { + t.Fatal(err) + } + if result.IsError { + t.Fatalf("unexpected error from run_stop: %v", result) + } + if !stopped { + t.Fatal("expected the stop function to have been invoked") + } + + // A second run_stop for the same, now-inactive path is idempotent and + // must succeed rather than error. + result, err = client.CallTool(t.Context(), &mcp.CallToolParams{ + Name: "run_stop", + Arguments: map[string]any{"path": path}, + }) + if err != nil { + t.Fatal(err) + } + if result.IsError { + t.Fatalf("expected second run_stop for the same path to succeed idempotently, got error: %v", result) + } +} diff --git a/pkg/mcp/tools_run_test.go b/pkg/mcp/tools_run_test.go new file mode 100644 index 0000000000..772dcb1c54 --- /dev/null +++ b/pkg/mcp/tools_run_test.go @@ -0,0 +1,164 @@ +package mcp + +import ( + "context" + "fmt" + "strings" + "testing" + + "github.com/modelcontextprotocol/go-sdk/mcp" + "knative.dev/func/pkg/mcp/mock" +) + +// TestTool_Run_Args ensures the run tool passes the correct arguments to the +// process starter and returns pid/url from the started process. +func TestTool_Run_Args(t *testing.T) { + path := testAbsPath("my-func") + + starter := mock.NewProcessStarter() + starter.StartFn = func(ctx context.Context, subcommand string, args ...string) (int, string, string, func() error, error) { + if subcommand != "run" { + t.Fatalf("expected subcommand 'run', got %q", subcommand) + } + + // path, --json, registry, build, port -> 3 string flags (path, registry, address) * 2 + 1 bool-ish (--json) + 1 (--build=true, standalone flag) + wantArgs := map[string]string{ + "--path": path, + "--registry": "ghcr.io/user", + "--address": "127.0.0.1:9090", + } + got := argsToMap(args) + for flag, val := range wantArgs { + if got[flag] != val { + t.Fatalf("expected %s=%s, got args %v", flag, val, args) + } + } + if _, ok := got["--json"]; !ok { + t.Fatalf("expected --json flag, got args %v", args) + } + if _, ok := got["--build=true"]; !ok { + t.Fatalf("expected --build=true flag, got args %v", args) + } + + return 4242, "127.0.0.1", "9090", func() error { return nil }, nil + } + + client, server, err := newTestPair(t, WithProcessStarter(starter)) + if err != nil { + t.Fatal(err) + } + server.readonly.Store(false) + + port := 9090 + result, err := client.CallTool(t.Context(), &mcp.CallToolParams{ + Name: "run", + Arguments: map[string]any{ + "path": path, + "registry": "ghcr.io/user", + "build": true, + "port": port, + }, + }) + if err != nil { + t.Fatal(err) + } + if result.IsError { + t.Fatalf("unexpected error result: %v", result) + } + if !starter.StartInvoked { + t.Fatal("process starter was not invoked") + } + + text := resultToString(result) + if wantURL := "http://127.0.0.1:9090"; !strings.Contains(text, wantURL) { + t.Fatalf("expected result to contain %q, got %q", wantURL, text) + } + if !strings.Contains(text, "4242") { + t.Fatalf("expected result to contain pid 4242, got %q", text) + } +} + +// TestTool_Run_AllowedInReadonly ensures the run tool is NOT gated by +// readonly mode: it is a local-only operation (no cluster mutation), unlike +// deploy/delete. +func TestTool_Run_AllowedInReadonly(t *testing.T) { + starter := mock.NewProcessStarter() + starter.StartFn = func(ctx context.Context, subcommand string, args ...string) (int, string, string, func() error, error) { + return 1, "127.0.0.1", "8080", func() error { return nil }, nil + } + + client, _, err := newTestPairCore(t, true, WithProcessStarter(starter)) + if err != nil { + t.Fatal(err) + } + + result, err := client.CallTool(t.Context(), &mcp.CallToolParams{ + Name: "run", + Arguments: map[string]any{"path": testAbsPath("my-func")}, + }) + if err != nil { + t.Fatal(err) + } + if result.IsError { + t.Fatalf("expected run to be allowed in readonly mode, got error: %v", result) + } +} + +// TestTool_Run_Duplicate ensures a second run for the same path is rejected +// while the first is still active. +func TestTool_Run_Duplicate(t *testing.T) { + starter := mock.NewProcessStarter() + starter.StartFn = func(ctx context.Context, subcommand string, args ...string) (int, string, string, func() error, error) { + return 1, "127.0.0.1", "8080", func() error { return nil }, nil + } + + client, server, err := newTestPair(t, WithProcessStarter(starter)) + if err != nil { + t.Fatal(err) + } + server.readonly.Store(false) + + params := &mcp.CallToolParams{Name: "run", Arguments: map[string]any{"path": testAbsPath("dup-func")}} + + result, err := client.CallTool(t.Context(), params) + if err != nil { + t.Fatal(err) + } + if result.IsError { + t.Fatalf("unexpected error on first run: %v", result) + } + + result, err = client.CallTool(t.Context(), params) + if err != nil { + t.Fatal(err) + } + if !result.IsError { + t.Fatal("expected second run for the same path to be rejected") + } +} + +// TestTool_Run_StartError ensures a failure from the process starter is +// surfaced as a tool error. +func TestTool_Run_StartError(t *testing.T) { + starter := mock.NewProcessStarter() + starter.StartFn = func(ctx context.Context, subcommand string, args ...string) (int, string, string, func() error, error) { + return 0, "", "", nil, fmt.Errorf("boom") + } + + client, server, err := newTestPair(t, WithProcessStarter(starter)) + if err != nil { + t.Fatal(err) + } + server.readonly.Store(false) + + result, err := client.CallTool(t.Context(), &mcp.CallToolParams{ + Name: "run", + Arguments: map[string]any{"path": testAbsPath("err-func")}, + }) + if err != nil { + t.Fatal(err) + } + if !result.IsError { + t.Fatal("expected run to surface the process starter's error") + } +} diff --git a/pkg/mcp/tools_test.go b/pkg/mcp/tools_test.go index a4931ecdbf..a52696165a 100644 --- a/pkg/mcp/tools_test.go +++ b/pkg/mcp/tools_test.go @@ -1,10 +1,19 @@ package mcp import ( + "os" + "path/filepath" "strings" "testing" ) +// testAbsPath returns a platform-native absolute path for name, suitable for +// tools which require an absolute path. A POSIX literal such as "/tmp/f" can +// not be used directly because it is not absolute on Windows. +func testAbsPath(name string) string { + return filepath.Join(os.TempDir(), name) +} + // validateArgLength validates that the args slice has the expected length based on // the number of string flags (2 args each: flag + value) and bool flags (1 arg each). func validateArgLength(t *testing.T, args []string, stringFlagsCount, boolFlagsCount int) {