diff --git a/README.md b/README.md index 4d360b0..7f2c71a 100644 --- a/README.md +++ b/README.md @@ -47,6 +47,7 @@ somewhere that is **not** the managed cluster (PRD §13 #10). ```sh medea serve \ --listen 0.0.0.0:7600 \ + --mcp-listen 127.0.0.1:7601 \ --store /var/lib/medea/medea.db \ --creds-dir /var/lib/medea/creds \ --token-file /etc/medea/token \ @@ -64,9 +65,38 @@ Notes: the server read-only. - **Credentials** (talosconfig/kubeconfig) live under `--creds-dir`, **not** in the bbolt store, and are never exported. See [Credentials](#credentials). +- **MCP** is disabled unless `--mcp-listen` is set. It serves a curated, + read-only Streamable HTTP endpoint at `/mcp`; bind it only to loopback or a + trusted private network. It does not expose credential retrieval or mutation + tools. See [`design/mcp.md`](design/mcp.md). For a systemd unit and a container image, see [`deploy/`](deploy/). +## Iris MCP integration + +Start Medea with an MCP listener reachable from Iris, then add the server to +Iris policy: + +```yaml +mcpServers: + medea: + url: http://medea.example.internal:7601/mcp + users: [martin] + tools: + - list_clusters + - get_cluster + - list_node_pools + - list_machines + - list_hosts + - get_rollout + - list_rollouts + approval: never +``` + +The endpoint currently has no application-layer authentication. Keep it on a +private interface or place it behind an authenticated TLS reverse proxy. Its +catalog is read-only, but it still reveals cluster inventory and topology. + ## Credentials Per managed cluster, drop the admin credentials under the creds dir (mode 0600, diff --git a/cmd/medea/serve.go b/cmd/medea/serve.go index 76e1630..5825eea 100644 --- a/cmd/medea/serve.go +++ b/cmd/medea/serve.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "net" + "net/http" "os" "os/signal" "strings" @@ -19,6 +20,7 @@ import ( pb "github.com/crunchloop/medea/gen/medea/v1" "github.com/crunchloop/medea/internal/auth" "github.com/crunchloop/medea/internal/creds" + "github.com/crunchloop/medea/internal/delivery/mcpapi" "github.com/crunchloop/medea/internal/provision" "github.com/crunchloop/medea/internal/provision/matchbox" "github.com/crunchloop/medea/internal/refresh" @@ -87,6 +89,7 @@ var ( serveBootstrap bool serveBootstrapIntv time.Duration + serveMCPListen string ) func init() { @@ -120,6 +123,7 @@ func init() { f.DurationVar(&serveProvisionTO, "provision-timeout", 20*time.Minute, "how long a host may stay provisioning before it is marked failed") f.BoolVar(&serveBootstrap, "bootstrap", false, "enable the cluster-bootstrap executor (Medea-driven new-cluster creation; default off)") f.DurationVar(&serveBootstrapIntv, "bootstrap-interval", 30*time.Second, "how often the bootstrap executor advances in-flight cluster creations") + f.StringVar(&serveMCPListen, "mcp-listen", "", "serve read-only MCP over Streamable HTTP at /mcp (disabled when empty)") rootCmd.AddCommand(serveCmd) } @@ -166,7 +170,8 @@ func runServe(_ *cobra.Command, _ []string) error { grpc.UnaryInterceptor(auth.UnaryInterceptor(token)), grpc.StreamInterceptor(auth.StreamInterceptor(token)), ) - pb.RegisterMedeaServer(srv, server.New(st, server.WithCreds(cs))) + api := server.New(st, server.WithCreds(cs)) + pb.RegisterMedeaServer(srv, api) lis, err := net.Listen("tcp", serveListen) if err != nil { @@ -224,15 +229,44 @@ func runServe(_ *cobra.Command, _ []string) error { fmt.Fprintln(os.Stderr, "medea: bootstrap executor disabled (pass --bootstrap to enable)") } - errCh := make(chan error, 1) + errCh := make(chan error, 2) go func() { errCh <- srv.Serve(lis) }() fmt.Fprintf(os.Stderr, "medea: serving on %s (store=%s, tls cert=%s)\n", serveListen, serveStore, serveCert) + var mcpHTTP *http.Server + if serveMCPListen != "" { + mcpServer, err := mcpapi.NewServer(api) + if err != nil { + return fmt.Errorf("MCP server: %w", err) + } + mux := http.NewServeMux() + mux.Handle("/mcp", mcpServer.Handler()) + mcpHTTP = &http.Server{ + Addr: serveMCPListen, + Handler: mux, + ReadHeaderTimeout: 10 * time.Second, + } + go func() { + err := mcpHTTP.ListenAndServe() + if !errors.Is(err, http.ErrServerClosed) { + errCh <- err + } + }() + fmt.Fprintf(os.Stderr, "medea: read-only MCP serving on http://%s/mcp\n", serveMCPListen) + } + select { case err := <-errCh: return err case <-ctx.Done(): fmt.Fprintln(os.Stderr, "medea: shutting down") + if mcpHTTP != nil { + shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := mcpHTTP.Shutdown(shutdownCtx); err != nil { + return fmt.Errorf("shut down MCP server: %w", err) + } + } srv.GracefulStop() return nil } diff --git a/cmd/medea/serve_mcp_test.go b/cmd/medea/serve_mcp_test.go new file mode 100644 index 0000000..c0174fd --- /dev/null +++ b/cmd/medea/serve_mcp_test.go @@ -0,0 +1,133 @@ +package main + +import ( + "bytes" + "context" + "net" + "os" + "os/exec" + "path/filepath" + "testing" + "time" + + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +// TestServeMCPProcess exercises the composition root rather than only the MCP +// package: a real Medea child process generates TLS, opens bbolt, starts gRPC +// and /mcp, serves an SDK client, and shuts both listeners down on interrupt. +func TestServeMCPProcess(t *testing.T) { + if os.Getenv("MEDEA_TEST_SERVE_MCP_HELPER") == "1" { + runServeMCPHelper(t) + return + } + + grpcAddr := reserveAddress(t) + mcpAddr := reserveAddress(t) + dir := t.TempDir() + cmd := exec.Command(os.Args[0], "-test.run=^TestServeMCPProcess$") + cmd.Env = append(os.Environ(), + "MEDEA_TEST_SERVE_MCP_HELPER=1", + "MEDEA_TEST_GRPC_ADDR="+grpcAddr, + "MEDEA_TEST_MCP_ADDR="+mcpAddr, + "MEDEA_TEST_DIR="+dir, + ) + var output bytes.Buffer + cmd.Stdout = &output + cmd.Stderr = &output + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { done <- cmd.Wait() }() + + endpoint := "http://" + mcpAddr + "/mcp" + session := waitForMCP(t, endpoint, done, &output) + listed, err := session.ListTools(context.Background(), nil) + if err != nil { + _ = cmd.Process.Kill() + t.Fatalf("list tools from child: %v\n%s", err, output.String()) + } + if len(listed.Tools) != 7 { + _ = cmd.Process.Kill() + t.Fatalf("child advertised %d tools, want 7\n%s", len(listed.Tools), output.String()) + } + result, err := session.CallTool(context.Background(), &mcp.CallToolParams{Name: "list_clusters"}) + if err != nil || result.IsError { + _ = cmd.Process.Kill() + t.Fatalf("call list_clusters through child: result=%+v err=%v\n%s", result, err, output.String()) + } + if err := session.Close(); err != nil { + _ = cmd.Process.Kill() + t.Fatal(err) + } + + if err := cmd.Process.Signal(os.Interrupt); err != nil { + _ = cmd.Process.Kill() + t.Fatal(err) + } + select { + case err := <-done: + if err != nil { + t.Fatalf("Medea child exited with error: %v\n%s", err, output.String()) + } + case <-time.After(15 * time.Second): + _ = cmd.Process.Kill() + t.Fatalf("Medea child did not shut down gracefully\n%s", output.String()) + } +} + +func runServeMCPHelper(t *testing.T) { + dir := os.Getenv("MEDEA_TEST_DIR") + serveListen = os.Getenv("MEDEA_TEST_GRPC_ADDR") + serveMCPListen = os.Getenv("MEDEA_TEST_MCP_ADDR") + serveStore = filepath.Join(dir, "medea.db") + serveToken = "test-token" + serveTokenFile = "" + serveCert = filepath.Join(dir, "cert.pem") + serveKey = filepath.Join(dir, "key.pem") + serveCredsDir = filepath.Join(dir, "creds") + serveCredsBackend = "file" + serveRefreshIntv = time.Hour + serveRollouts = false + serveProvisioning = false + serveBootstrap = false + if err := runServe(nil, nil); err != nil { + t.Fatal(err) + } +} + +func reserveAddress(t *testing.T) string { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + addr := lis.Addr().String() + if err := lis.Close(); err != nil { + t.Fatal(err) + } + return addr +} + +func waitForMCP(t *testing.T, endpoint string, done <-chan error, output *bytes.Buffer) *mcp.ClientSession { + t.Helper() + deadline := time.Now().Add(15 * time.Second) + for time.Now().Before(deadline) { + select { + case err := <-done: + t.Fatalf("Medea child exited before MCP became ready: %v\n%s", err, output.String()) + default: + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + client := mcp.NewClient(&mcp.Implementation{Name: "medea-process-test", Version: "1"}, nil) + session, err := client.Connect(ctx, &mcp.StreamableClientTransport{Endpoint: endpoint}, nil) + cancel() + if err == nil { + return session + } + time.Sleep(100 * time.Millisecond) + } + t.Fatalf("MCP did not become ready at %s\n%s", endpoint, output.String()) + return nil +} diff --git a/design/README.md b/design/README.md index 4074f01..ee4dc52 100644 --- a/design/README.md +++ b/design/README.md @@ -19,6 +19,7 @@ reasoning stays useful after the work ships. | --- | --- | | [`datastore.md`](datastore.md) | The embedded (bbolt) store: desired-vs-observed schema, proto-as-storage, revisions, watch, concurrency, crash recovery, desired-state export. | | [`api-and-auth.md`](api-and-auth.md) | The gRPC service (intent verbs, thin watch), v1 auth (bearer token over TLS), and credential storage (separate from bbolt). Resolves #8/#9. | +| [`mcp.md`](mcp.md) | The Iris-facing read-only MCP adapter over the existing application API, Streamable HTTP boundary, curated tools, and the future SEP-2663 Tasks mapping for durable rollouts. | | [`credentials.md`](credentials.md) | **(Phase A, Draft for review)** Making Medea the durable, off-host owner of a cluster's credentials + machine-secrets bundle so `home-cluster`'s `_out/` can be deleted: a 1Password-backed `CredentialStore` (built, was planned in api-and-auth §5), a guarded `GetCredentials` RPC, and the cutover. Prerequisite for Phase B (`cluster-bootstrap.md`). | | [`talos-client.md`](talos-client.md) | Talos & kube clients (no shelling): small Medea interfaces over the `machinery` client (OS upgrade/snapshot/health/version) + client-go (readiness/drain); the quarantined main-module `upgrade-k8s`; installer-image/schematic derivation. | | [`rollout-controller.md`](rollout-controller.md) | The v1 version-rollout reconciler: per-node state machine, OS vs K8s paths, halt-on-failure, resume-after-reboot, control-plane snapshot safety. (Trigger in §1 superseded by `rollout-safety.md`.) | diff --git a/design/mcp.md b/design/mcp.md new file mode 100644 index 0000000..bcc09fe --- /dev/null +++ b/design/mcp.md @@ -0,0 +1,149 @@ +# MCP delivery adapter + +**Status:** Implemented (read-only surface); task-backed operations deferred +**Date:** 2026-07-15 + +Scope: the Model Context Protocol adapter Iris uses to inspect Medea, its +transport and security boundary, and how long-running Medea operations should +map to MCP Tasks once the Go ecosystem supports the extension. + +## 1. Shape + +MCP is a peer delivery adapter beside gRPC. It does not read bbolt directly and +does not define another application service: + +```text +Iris ── Streamable HTTP ──> internal/delivery/mcpapi + │ + v + internal/server.Server + │ + v + store.Store +``` + +`internal/server.Server` satisfies the small `mcpapi.Queries` interface. This +keeps all validation and read semantics on the same path as the CLI and other +gRPC clients. The official MCP Go SDK is confined to `internal/delivery/mcpapi`. + +This follows Ariadne's delivery posture: typed tool arguments, a curated tool +suite, structured results, compact text fallbacks, stable structured errors, +and adapter-level contract tests. + +## 2. Curated tool surface + +The first surface is intentionally read-only: + +| Tool | Medea operation | +| --- | --- | +| `list_clusters` | `ListClusters` | +| `get_cluster` | `GetCluster` | +| `list_node_pools` | `ListNodePools` | +| `list_machines` | `ListMachines` | +| `list_hosts` | `ListHosts` | +| `get_rollout` | `GetRollout` | +| `list_rollouts` | `ListRollouts` | + +Every tool advertises `readOnlyHint`, `idempotentHint`, and a closed-world +`openWorldHint`. Protobuf responses are encoded with protobuf JSON semantics, +snake_case field names, and symbolic enum values. `structuredContent` is the +lossless result; text content only identifies the result and reports counts. + +Credential retrieval is not exposed. Neither are cluster creation/deletion, +host mutations, desired-version writes, safety-gate changes, or rollout +controls. Raw store access and arbitrary query tools are also out of scope. + +## 3. Transport and security + +`medea serve --mcp-listen
` enables Streamable HTTP at `/mcp`; the +empty default disables it. The endpoint has no application-layer +authentication in this first version because Iris's current remote MCP client +supports anonymous or OAuth connections, while Medea's existing API uses a +static gRPC bearer token. + +Consequences: + +- the MCP endpoint is read-only and cannot return stored credentials; +- bind it to loopback, a private sidecar network, or another trusted interface; +- for a shared network, put it behind an authenticated TLS reverse proxy and + network policy; +- do not add mutations until Medea and Iris share an authenticated MCP + authorization context. Iris approval is a human policy gate, not network + authentication and not a replacement for Medea's rollout safety chain. + +MCP and gRPC use separate listeners because gRPC is TLS-only today and the MCP +Go SDK owns an HTTP handler. Converging them behind one reverse proxy is a +deployment concern, not an adapter concern. + +## 4. Errors + +Domain/application failures are tool execution errors so a model can inspect +and correct its arguments. They use stable codes: + +| gRPC status | MCP problem code | +| --- | --- | +| `InvalidArgument` | `invalid_request` | +| `NotFound` | `not_found` | +| `FailedPrecondition` | `failed_precondition` | +| `Aborted` | `conflict` | +| `Canceled`, `DeadlineExceeded` | `query_canceled` | +| other | `query_unavailable` (detail redacted) | + +## 5. MCP Tasks and Medea operations + +Tasks are useful, but not for the read tools above. The strong fit is an +eventual `create_rollout` MCP tool: + +1. The synchronous Medea command validates the full safety chain and durably + creates the rollout job. +2. When Iris and Medea negotiate the Tasks extension, the MCP call returns that + durable operation as a task handle. +3. `tasks/get` projects the persisted rollout job; store watch events can later + drive task-status notifications without changing the source of truth. +4. A non-Tasks client receives the normal immediate "job accepted" result. + +The current Tasks extension is [SEP-2663](https://modelcontextprotocol.io/seps/2663-tasks-extension), +which supersedes the experimental core Tasks API from MCP 2025-11-25. It uses +extension negotiation, server-directed task creation, `tasks/get`, +`tasks/update`, and `tasks/cancel`; terminal results are returned inline by +`tasks/get`. As of this record, Tasks remain on the official Go SDK roadmap, so +Medea does not implement a private copy of the wire protocol. + +The future rollout projection is: + +| Medea rollout job | MCP task | +| --- | --- | +| `PENDING`, `RUNNING` | `working` | +| `PAUSED` | `working`, with a paused status message | +| `DONE` | `completed`, with the final rollout result | +| `FAILED` | `completed`, with `CallToolResult.isError=true` (an application outcome, not a JSON-RPC failure) | +| `CANCELLED` | `cancelled` | + +`PAUSED` must not become `input_required` unless Medea actually emits a bounded +request that `tasks/update` can satisfy. A separate operator may resume a +rollout; merely being paused is not necessarily a question for the task caller. + +### Persistence prerequisite + +Medea currently keys one rollout job by `(cluster, pool)` and overwrites it on a +later rollout. That is enough for the v1 executor but not for MCP's durable task +retention: an old task ID must continue resolving for its TTL after a newer job +starts. Before enabling Tasks, the Rollout aggregate needs a stable job ID and +retained terminal history (or an equivalently durable projection). A second, +MCP-only task database is rejected because it would duplicate lifecycle truth. + +Cluster bootstrap is another potential task-backed tool, but it first needs a +read application operation for bootstrap status and the same durable identity +analysis. It is not exposed by reaching around the application boundary into +the store. + +## 6. Decisions + +1. MCP is a delivery adapter over `internal/server`, not a store client. +2. The initial catalog is curated and read-only; credentials and mutations are + absent by construction. +3. Streamable HTTP is used because it is the transport Iris consumes. +4. Structured protobuf JSON is canonical; text is a compact fallback. +5. Rollouts should use SEP-2663 Tasks when client and SDK support lands, backed + by the Rollout aggregate after it gains durable identity/history. +6. No proprietary Tasks wire shim and no duplicate task store. diff --git a/go.mod b/go.mod index 3f0286b..99213f1 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.26.3 require ( github.com/1password/onepassword-sdk-go v0.4.1-0.20260605221002-f1117e36ce06 github.com/cosi-project/runtime v1.14.1 + github.com/modelcontextprotocol/go-sdk v1.6.1 github.com/siderolabs/go-kubernetes v0.2.38 github.com/siderolabs/talos v1.13.5 github.com/siderolabs/talos/pkg/machinery v1.13.5 @@ -71,6 +72,7 @@ require ( github.com/google/gnostic-models v0.7.1 // indirect github.com/google/go-cmp v0.7.0 // indirect github.com/google/go-containerregistry v0.21.5 // indirect + github.com/google/jsonschema-go v0.4.3 // indirect github.com/google/uuid v1.6.0 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 // indirect github.com/hashicorp/errwrap v1.1.0 // indirect @@ -105,6 +107,8 @@ require ( github.com/russross/blackfriday/v2 v2.1.0 // indirect github.com/ryanuber/go-glob v1.0.0 // indirect github.com/sasha-s/go-deadlock v0.3.6 // indirect + github.com/segmentio/asm v1.1.3 // indirect + github.com/segmentio/encoding v0.5.4 // indirect github.com/siderolabs/crypto v0.6.5 // indirect github.com/siderolabs/gen v0.8.6 // indirect github.com/siderolabs/go-api-signature v0.3.12 // indirect @@ -120,6 +124,7 @@ require ( github.com/vbatts/tar-split v0.12.2 // indirect github.com/x448/float16 v0.8.4 // indirect github.com/xlab/treeprint v1.2.0 // indirect + github.com/yosida95/uritemplate/v3 v3.0.2 // indirect go.opentelemetry.io/proto/otlp v1.9.0 // indirect go.uber.org/multierr v1.11.0 // indirect go.uber.org/zap v1.27.1 // indirect diff --git a/go.sum b/go.sum index 0c32d53..58f8a71 100644 --- a/go.sum +++ b/go.sum @@ -127,6 +127,8 @@ github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1v github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8= github.com/gobwas/glob v0.2.3 h1:A4xDbljILXROh+kObIiy5kIaPYD8e96x1tgBhUI5J+Y= github.com/gobwas/glob v0.2.3/go.mod h1:d3Ez4x06l9bZtSvzIay5+Yzi0fmZzPgnTbPcKjJAkT8= +github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY= +github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= github.com/google/btree v1.1.3 h1:CVpQJjYgC4VbzxeGVHfvZrv1ctoYCAI8vbl07Fcxlyg= @@ -140,6 +142,8 @@ github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX github.com/google/go-containerregistry v0.21.5 h1:KTJG9Pn/jC0VdZR6ctV3/jcN+q6/Iqlx0sTVz3ywZlM= github.com/google/go-containerregistry v0.21.5/go.mod h1:ySvMuiWg+dOsRW0Hw8GYwfMwBlNRTmpYBFJPlkco5zU= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/google/jsonschema-go v0.4.3 h1:/DBOLZTfDow7pe2GmaJNhltueGTtDKICi8V8p+DQPd0= +github.com/google/jsonschema-go v0.4.3/go.mod h1:r5quNTdLOYEz95Ru18zA0ydNbBuYoo9tgaYcxEYhJVE= github.com/google/pprof v0.0.0-20260115054156-294ebfa9ad83 h1:z2ogiKUYzX5Is6zr/vP9vJGqPwcdqsWjOt+V8J7+bTc= github.com/google/pprof v0.0.0-20260115054156-294ebfa9ad83/go.mod h1:MxpfABSjhmINe3F1It9d+8exIHFvUqtLIRCdOGNXqiI= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= @@ -183,6 +187,8 @@ github.com/moby/moby/api v1.54.1 h1:TqVzuJkOLsgLDDwNLmYqACUuTehOHRGKiPhvH8V3Nn4= github.com/moby/moby/api v1.54.1/go.mod h1:+RQ6wluLwtYaTd1WnPLykIDPekkuyD/ROWQClE83pzs= github.com/moby/term v0.5.2 h1:6qk3FJAFDs6i/q3W/pQ97SX192qKfZgGjCQqfCJkgzQ= github.com/moby/term v0.5.2/go.mod h1:d3djjFCrjnB+fl8NJux+EJzu0msscUP+f8it8hPkFLc= +github.com/modelcontextprotocol/go-sdk v1.6.1 h1:0zOSupjKUxPKSocPT1Wtago+mUHU2/uZ4xSOY0FGReU= +github.com/modelcontextprotocol/go-sdk v1.6.1/go.mod h1:kzm3kzFL1/+AziGOE0nUs3gvPoNxMCvkxokMkuFapXQ= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= @@ -237,6 +243,10 @@ github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 h1:KRzFb2m7YtdldCEkzs6KqmJw4nqEV github.com/santhosh-tekuri/jsonschema/v6 v6.0.2/go.mod h1:JXeL+ps8p7/KNMjDQk3TCwPpBy0wYklyWTfbkIzdIFU= github.com/sasha-s/go-deadlock v0.3.6 h1:TR7sfOnZ7x00tWPfD397Peodt57KzMDo+9Ae9rMiUmw= github.com/sasha-s/go-deadlock v0.3.6/go.mod h1:CUqNyyvMxTyjFqDT7MRg9mb4Dv/btmGTqSR+rky/UXo= +github.com/segmentio/asm v1.1.3 h1:WM03sfUOENvvKexOLp+pCqgb/WDjsi7EK8gIsICtzhc= +github.com/segmentio/asm v1.1.3/go.mod h1:Ld3L4ZXGNcSLRg4JBsZ3//1+f/TjYl0Mzen/DQy1EJg= +github.com/segmentio/encoding v0.5.4 h1:OW1VRern8Nw6ITAtwSZ7Idrl3MXCFwXHPgqESYfvNt0= +github.com/segmentio/encoding v0.5.4/go.mod h1:HS1ZKa3kSN32ZHVZ7ZLPLXWvOVIiZtyJnO1gPH1sKt0= github.com/sergi/go-diff v1.4.0 h1:n/SP9D5ad1fORl+llWyN+D6qoUETXNZARKjyY2/KVCw= github.com/sergi/go-diff v1.4.0/go.mod h1:A0bzQcvG0E7Rwjx0REVgAGH58e96+X0MeOfepqsbeW4= github.com/siderolabs/crypto v0.6.5 h1:Elq5tpWP2ApZ4Y+Kg+eIDiiWbmriCPI1mjYIMwvsYkw= @@ -290,6 +300,8 @@ github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= github.com/xlab/treeprint v1.2.0 h1:HzHnuAF1plUN2zGlAFHbSQP2qJ0ZAD3XF5XD7OesXRQ= github.com/xlab/treeprint v1.2.0/go.mod h1:gj5Gd3gPdKtR1ikdDK6fnFLdmIS0X30kTTuNd/WEJu0= +github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4= +github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4= github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= go.etcd.io/bbolt v1.5.0 h1:S7GAl7Fxv12yohbwFfIbQCGDWbQbtDGPET4P/bD4lxU= go.etcd.io/bbolt v1.5.0/go.mod h1:mkltfYE5aUHQxUct9N9V+Kp7aSjFqjgrhcXIS70Lrdk= diff --git a/internal/delivery/mcpapi/problem.go b/internal/delivery/mcpapi/problem.go new file mode 100644 index 0000000..eb64c8e --- /dev/null +++ b/internal/delivery/mcpapi/problem.go @@ -0,0 +1,43 @@ +package mcpapi + +import ( + "encoding/json" + + "github.com/modelcontextprotocol/go-sdk/mcp" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +type Problem struct { + Code string `json:"code"` + Detail string `json:"detail"` +} + +func errorResult(err error) *mcp.CallToolResult { + problem := mapError(err) + encoded, marshalErr := json.Marshal(problem) + if marshalErr != nil { + encoded = []byte(`{"code":"query_unavailable","detail":"Medea query is temporarily unavailable"}`) + } + return &mcp.CallToolResult{ + IsError: true, StructuredContent: problem, + Content: []mcp.Content{&mcp.TextContent{Text: string(encoded)}}, + } +} + +func mapError(err error) Problem { + switch status.Code(err) { + case codes.InvalidArgument: + return Problem{Code: "invalid_request", Detail: err.Error()} + case codes.NotFound: + return Problem{Code: "not_found", Detail: err.Error()} + case codes.FailedPrecondition: + return Problem{Code: "failed_precondition", Detail: err.Error()} + case codes.Aborted: + return Problem{Code: "conflict", Detail: err.Error()} + case codes.Canceled, codes.DeadlineExceeded: + return Problem{Code: "query_canceled", Detail: err.Error()} + default: + return Problem{Code: "query_unavailable", Detail: "Medea query is temporarily unavailable"} + } +} diff --git a/internal/delivery/mcpapi/server.go b/internal/delivery/mcpapi/server.go new file mode 100644 index 0000000..f675a97 --- /dev/null +++ b/internal/delivery/mcpapi/server.go @@ -0,0 +1,51 @@ +// Package mcpapi exposes Medea's read-only Model Context Protocol delivery +// adapter. MCP SDK types are deliberately confined to this package. +package mcpapi + +import ( + "context" + "errors" + "net/http" + + "github.com/modelcontextprotocol/go-sdk/mcp" + + pb "github.com/crunchloop/medea/gen/medea/v1" +) + +const ServerVersion = "1.0.0" + +// Queries is the existing Medea application read surface consumed by MCP. +// internal/server.Server satisfies this interface; MCP does not read the store +// directly or introduce a second application API. +type Queries interface { + GetCluster(context.Context, *pb.GetClusterRequest) (*pb.Cluster, error) + ListClusters(context.Context, *pb.ListClustersRequest) (*pb.ListClustersResponse, error) + ListNodePools(context.Context, *pb.ListNodePoolsRequest) (*pb.ListNodePoolsResponse, error) + ListMachines(context.Context, *pb.ListMachinesRequest) (*pb.ListMachinesResponse, error) + ListHosts(context.Context, *pb.ListHostsRequest) (*pb.ListHostsResponse, error) + GetRollout(context.Context, *pb.GetRolloutRequest) (*pb.GetRolloutResponse, error) + ListRollouts(context.Context, *pb.ListRolloutsRequest) (*pb.ListRolloutsResponse, error) +} + +// Server owns the MCP protocol server while keeping transport types out of the +// composition root. +type Server struct { + server *mcp.Server +} + +func NewServer(queries Queries) (*Server, error) { + if queries == nil { + return nil, errors.New("MCP server requires Medea queries") + } + server := mcp.NewServer(&mcp.Implementation{Name: "medea", Version: ServerVersion}, nil) + registerTools(server, queries) + return &Server{server: server}, nil +} + +// Handler serves MCP over Streamable HTTP. The handler is stateful so clients +// may negotiate ordinary MCP sessions; tool execution itself remains stateless. +func (s *Server) Handler() http.Handler { + return mcp.NewStreamableHTTPHandler(func(*http.Request) *mcp.Server { + return s.server + }, nil) +} diff --git a/internal/delivery/mcpapi/server_test.go b/internal/delivery/mcpapi/server_test.go new file mode 100644 index 0000000..32af75c --- /dev/null +++ b/internal/delivery/mcpapi/server_test.go @@ -0,0 +1,346 @@ +package mcpapi_test + +import ( + "context" + "net/http/httptest" + "path/filepath" + "slices" + "strings" + "testing" + + "github.com/modelcontextprotocol/go-sdk/mcp" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + pb "github.com/crunchloop/medea/gen/medea/v1" + "github.com/crunchloop/medea/internal/delivery/mcpapi" + "github.com/crunchloop/medea/internal/server" + "github.com/crunchloop/medea/internal/store" +) + +func TestReadOnlyCatalogAndStructuredResults(t *testing.T) { + session, st := connect(t) + seed(t, st) + + listed, err := session.ListTools(context.Background(), nil) + if err != nil { + t.Fatal(err) + } + var names []string + for _, tool := range listed.Tools { + names = append(names, tool.Name) + if tool.Annotations == nil || !tool.Annotations.ReadOnlyHint || !tool.Annotations.IdempotentHint { + t.Fatalf("tool %q is not annotated read-only and idempotent: %+v", tool.Name, tool.Annotations) + } + } + slices.Sort(names) + want := []string{"get_cluster", "get_rollout", "list_clusters", "list_hosts", "list_machines", "list_node_pools", "list_rollouts"} + if !slices.Equal(names, want) { + t.Fatalf("tools = %v, want %v", names, want) + } + + assertRequiredInputs(t, listed.Tools) + + tests := []struct { + name string + arguments map[string]any + validate func(*testing.T, map[string]any) + }{ + {"list_clusters", nil, func(t *testing.T, got map[string]any) { + if len(objectSlice(t, got, "clusters")) != 1 { + t.Fatalf("clusters = %#v", got) + } + }}, + {"get_cluster", map[string]any{"cluster": "home"}, func(t *testing.T, got map[string]any) { + if got["name"] != "home" || got["mode"] != "CLUSTER_MODE_MANUAL" { + t.Fatalf("cluster = %#v", got) + } + }}, + {"list_node_pools", map[string]any{"cluster": "home"}, func(t *testing.T, got map[string]any) { + pools := objectSlice(t, got, "node_pools") + if len(pools) != 2 { + t.Fatalf("node_pools = %#v", pools) + } + }}, + {"list_machines", map[string]any{"cluster": "home", "pool": "workers"}, func(t *testing.T, got map[string]any) { + machines := objectSlice(t, got, "machines") + if len(machines) != 1 || machines[0]["pool"] != "workers" || machines[0]["observed"].(map[string]any)["phase"] != "MACHINE_PHASE_READY" { + t.Fatalf("filtered machines = %#v", machines) + } + }}, + {"list_hosts", map[string]any{"cluster": "home", "pool": "workers"}, func(t *testing.T, got map[string]any) { + hosts := objectSlice(t, got, "hosts") + if len(hosts) != 1 || hosts[0]["mac"] != "aa:bb:cc:dd:ee:01" { + t.Fatalf("filtered hosts = %#v", hosts) + } + }}, + {"get_rollout", map[string]any{"cluster": "home", "pool": "workers"}, func(t *testing.T, got map[string]any) { + if got["cluster_rollout"].(map[string]any)["phase"] != "CLUSTER_ROLLOUT_PHASE_UPGRADING" { + t.Fatalf("cluster rollout = %#v", got) + } + rollouts := objectSlice(t, got, "machine_rollouts") + if len(rollouts) != 1 || rollouts[0]["state"] != "ROLLOUT_STATE_WAITING_HEALTHY" { + t.Fatalf("machine rollouts = %#v", rollouts) + } + }}, + {"list_rollouts", map[string]any{"cluster": "home"}, func(t *testing.T, got map[string]any) { + rollouts := objectSlice(t, got, "rollouts") + if len(rollouts) != 1 || rollouts[0]["state"] != "ROLLOUT_JOB_STATE_RUNNING" { + t.Fatalf("rollout jobs = %#v", rollouts) + } + }}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + result := call(t, session, tc.name, tc.arguments) + if result.IsError { + t.Fatalf("tool error: %+v", result.StructuredContent) + } + structured, ok := result.StructuredContent.(map[string]any) + if !ok { + t.Fatalf("structuredContent type = %T", result.StructuredContent) + } + tc.validate(t, structured) + text := result.Content[0].(*mcp.TextContent).Text + if !strings.Contains(text, "Use structuredContent") || strings.Contains(text, "kubernetes_version") { + t.Fatalf("text fallback is not compact: %q", text) + } + }) + } +} + +func TestDomainErrorsAreToolResults(t *testing.T) { + session, _ := connect(t) + result := call(t, session, "get_cluster", map[string]any{"cluster": "missing"}) + if !result.IsError { + t.Fatal("missing cluster did not return a tool execution error") + } + problem, ok := result.StructuredContent.(map[string]any) + if !ok || problem["code"] != "not_found" { + t.Fatalf("problem = %#v", result.StructuredContent) + } +} + +func TestRequiredArgumentsAreRejected(t *testing.T) { + session, _ := connect(t) + for _, name := range []string{"get_cluster", "list_node_pools", "list_machines", "list_hosts", "get_rollout", "list_rollouts"} { + t.Run(name, func(t *testing.T) { + result := call(t, session, name, map[string]any{}) + if !result.IsError { + t.Fatal("missing cluster argument was accepted") + } + }) + } +} + +func TestErrorMappingAndRedaction(t *testing.T) { + queries := errorQueries{} + session := connectQueries(t, queries) + tests := []struct { + cluster string + code string + }{ + {"invalid", "invalid_request"}, + {"missing", "not_found"}, + {"precondition", "failed_precondition"}, + {"conflict", "conflict"}, + {"canceled", "query_canceled"}, + {"deadline", "query_canceled"}, + {"internal", "query_unavailable"}, + } + for _, tc := range tests { + t.Run(tc.cluster, func(t *testing.T) { + result := call(t, session, "get_cluster", map[string]any{"cluster": tc.cluster}) + if !result.IsError { + t.Fatal("error was returned as success") + } + problem := result.StructuredContent.(map[string]any) + if problem["code"] != tc.code { + t.Fatalf("problem = %#v", problem) + } + if tc.cluster == "internal" && strings.Contains(problem["detail"].(string), "database password") { + t.Fatalf("internal detail leaked: %#v", problem) + } + }) + } +} + +func TestConcurrentSessionsAndReconnect(t *testing.T) { + st := mustStore(t) + url := startServerURL(t, server.New(st)) + seed(t, st) + + first := connectClient(t, url) + second := connectClient(t, url) + if call(t, first, "list_clusters", nil).IsError || call(t, second, "list_clusters", nil).IsError { + t.Fatal("concurrent session call failed") + } + if err := first.Close(); err != nil { + t.Fatal(err) + } + if call(t, second, "get_cluster", map[string]any{"cluster": "home"}).IsError { + t.Fatal("closing one session broke another") + } + reconnected := connectClient(t, url) + if call(t, reconnected, "list_rollouts", map[string]any{"cluster": "home"}).IsError { + t.Fatal("reconnected session call failed") + } +} + +func connect(t *testing.T) (*mcp.ClientSession, *store.BoltStore) { + t.Helper() + st := mustStore(t) + return connectClient(t, startServerURL(t, server.New(st))), st +} + +func connectQueries(t *testing.T, queries mcpapi.Queries) *mcp.ClientSession { + t.Helper() + return connectClient(t, startServerURL(t, queries)) +} + +func startServerURL(t *testing.T, queries mcpapi.Queries) string { + t.Helper() + srv, err := mcpapi.NewServer(queries) + if err != nil { + t.Fatal(err) + } + httpServer := httptest.NewServer(srv.Handler()) + t.Cleanup(httpServer.Close) + return httpServer.URL +} + +func connectClient(t *testing.T, endpoint string) *mcp.ClientSession { + t.Helper() + client := mcp.NewClient(&mcp.Implementation{Name: "medea-test", Version: "1"}, nil) + session, err := client.Connect(context.Background(), &mcp.StreamableClientTransport{Endpoint: endpoint}, nil) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = session.Close() }) + return session +} + +func seed(t *testing.T, st *store.BoltStore) { + t.Helper() + if _, err := st.PutClusterDesired(&pb.Cluster{ + Name: "home", Mode: pb.ClusterMode_CLUSTER_MODE_MANUAL, + Desired: &pb.ClusterDesired{TalosVersion: "v1.13.5", KubernetesVersion: "v1.36.2"}, + }, 0); err != nil { + t.Fatal(err) + } + for _, pool := range []*pb.NodePool{ + {Cluster: "home", Name: "workers", Role: pb.Role_ROLE_WORKER, Members: []string{"10.0.0.11"}}, + {Cluster: "home", Name: "controlplane", Role: pb.Role_ROLE_CONTROLPLANE, Members: []string{"10.0.0.10"}}, + } { + if _, err := st.PutNodePoolDesired(pool, 0); err != nil { + t.Fatal(err) + } + } + for _, machine := range []*pb.Machine{ + {Cluster: "home", Pool: "workers", TalosEndpoint: "10.0.0.11", Role: pb.Role_ROLE_WORKER}, + {Cluster: "home", Pool: "controlplane", TalosEndpoint: "10.0.0.10", Role: pb.Role_ROLE_CONTROLPLANE}, + } { + if _, err := st.PutMachineDesired(machine, 0); err != nil { + t.Fatal(err) + } + st.SetMachineObserved(machine.GetCluster(), machine.GetTalosEndpoint(), &pb.MachineObserved{ + Phase: pb.MachinePhase_MACHINE_PHASE_READY, TalosVersion: "v1.13.5", Healthy: true, + }) + } + for _, host := range []*pb.Host{ + {Cluster: "home", Pool: "workers", Mac: "aa:bb:cc:dd:ee:01", State: pb.HostState_HOST_STATE_READY}, + {Cluster: "home", Pool: "controlplane", Mac: "aa:bb:cc:dd:ee:02", State: pb.HostState_HOST_STATE_READY}, + } { + if _, err := st.PutHostDesired(host, 0); err != nil { + t.Fatal(err) + } + } + if err := st.PutClusterRollout(&pb.ClusterRollout{Cluster: "home", Phase: pb.ClusterRolloutPhase_CLUSTER_ROLLOUT_PHASE_UPGRADING}); err != nil { + t.Fatal(err) + } + if err := st.PutMachineRollout(&pb.MachineRollout{Cluster: "home", Addr: "10.0.0.11", State: pb.RolloutState_ROLLOUT_STATE_WAITING_HEALTHY}); err != nil { + t.Fatal(err) + } + if err := st.PutRolloutJob(&pb.Rollout{Cluster: "home", Pool: "workers", Kind: pb.RolloutKind_ROLLOUT_KIND_TALOS, State: pb.RolloutJobState_ROLLOUT_JOB_STATE_RUNNING}); err != nil { + t.Fatal(err) + } +} + +func mustStore(t *testing.T) *store.BoltStore { + t.Helper() + st, err := store.Open(filepath.Join(t.TempDir(), "medea.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = st.Close() }) + return st +} + +func call(t *testing.T, session *mcp.ClientSession, name string, arguments map[string]any) *mcp.CallToolResult { + t.Helper() + result, err := session.CallTool(context.Background(), &mcp.CallToolParams{Name: name, Arguments: arguments}) + if err != nil { + t.Fatalf("%s escaped as protocol error: %v", name, err) + } + return result +} + +func objectSlice(t *testing.T, object map[string]any, key string) []map[string]any { + t.Helper() + raw, ok := object[key].([]any) + if !ok { + t.Fatalf("%s type = %T in %#v", key, object[key], object) + } + out := make([]map[string]any, len(raw)) + for i := range raw { + out[i] = raw[i].(map[string]any) + } + return out +} + +func assertRequiredInputs(t *testing.T, tools []*mcp.Tool) { + t.Helper() + for _, tool := range tools { + schema := tool.InputSchema.(map[string]any) + required, _ := schema["required"].([]any) + if tool.Name == "list_clusters" { + if len(required) != 0 { + t.Fatalf("list_clusters required = %v", required) + } + continue + } + if len(required) != 1 || required[0] != "cluster" { + t.Fatalf("tool %q required = %v, want [cluster]", tool.Name, required) + } + } +} + +type errorQueries struct{} + +func (errorQueries) GetCluster(_ context.Context, req *pb.GetClusterRequest) (*pb.Cluster, error) { + codesByCluster := map[string]codes.Code{ + "invalid": codes.InvalidArgument, "missing": codes.NotFound, + "precondition": codes.FailedPrecondition, "conflict": codes.Aborted, + "canceled": codes.Canceled, "deadline": codes.DeadlineExceeded, + "internal": codes.Internal, + } + return nil, status.Error(codesByCluster[req.GetCluster()], "database password is hunter2") +} +func (errorQueries) ListClusters(context.Context, *pb.ListClustersRequest) (*pb.ListClustersResponse, error) { + panic("unexpected call") +} +func (errorQueries) ListNodePools(context.Context, *pb.ListNodePoolsRequest) (*pb.ListNodePoolsResponse, error) { + panic("unexpected call") +} +func (errorQueries) ListMachines(context.Context, *pb.ListMachinesRequest) (*pb.ListMachinesResponse, error) { + panic("unexpected call") +} +func (errorQueries) ListHosts(context.Context, *pb.ListHostsRequest) (*pb.ListHostsResponse, error) { + panic("unexpected call") +} +func (errorQueries) GetRollout(context.Context, *pb.GetRolloutRequest) (*pb.GetRolloutResponse, error) { + panic("unexpected call") +} +func (errorQueries) ListRollouts(context.Context, *pb.ListRolloutsRequest) (*pb.ListRolloutsResponse, error) { + panic("unexpected call") +} diff --git a/internal/delivery/mcpapi/tools.go b/internal/delivery/mcpapi/tools.go new file mode 100644 index 0000000..41ec68b --- /dev/null +++ b/internal/delivery/mcpapi/tools.go @@ -0,0 +1,96 @@ +package mcpapi + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/modelcontextprotocol/go-sdk/mcp" + "google.golang.org/protobuf/encoding/protojson" + "google.golang.org/protobuf/proto" + + pb "github.com/crunchloop/medea/gen/medea/v1" +) + +type emptyInput struct{} + +type clusterInput struct { + Cluster string `json:"cluster" jsonschema:"Medea cluster name"` +} + +type poolInput struct { + Cluster string `json:"cluster" jsonschema:"Medea cluster name"` + Pool string `json:"pool,omitempty" jsonschema:"Optional node-pool filter"` +} + +func registerTools(server *mcp.Server, queries Queries) { + mcp.AddTool(server, readTool("list_clusters", "List all clusters managed by Medea, including desired and observed versions, endpoints, revisions, and safety gates."), + func(ctx context.Context, _ *mcp.CallToolRequest, _ emptyInput) (*mcp.CallToolResult, any, error) { + result, err := queries.ListClusters(ctx, &pb.ListClustersRequest{}) + return protoResult(result, fmt.Sprintf("Medea manages %d clusters.", len(result.GetClusters())), err) + }) + + mcp.AddTool(server, readTool("get_cluster", "Get one Medea cluster by name, including desired and observed versions, endpoints, revision, and rollout/provisioning safety gates."), + func(ctx context.Context, _ *mcp.CallToolRequest, input clusterInput) (*mcp.CallToolResult, any, error) { + result, err := queries.GetCluster(ctx, &pb.GetClusterRequest{Cluster: input.Cluster}) + return protoResult(result, fmt.Sprintf("Retrieved cluster %q.", input.Cluster), err) + }) + + mcp.AddTool(server, readTool("list_node_pools", "List node pools for one Medea cluster, including members, desired Talos versions, rollout strategies, pause state, and provisioning selectors."), + func(ctx context.Context, _ *mcp.CallToolRequest, input clusterInput) (*mcp.CallToolResult, any, error) { + result, err := queries.ListNodePools(ctx, &pb.ListNodePoolsRequest{Cluster: input.Cluster}) + return protoResult(result, fmt.Sprintf("Found %d node pools in cluster %q.", len(result.GetNodePools()), input.Cluster), err) + }) + + mcp.AddTool(server, readTool("list_machines", "List machines and current observed health/version state for one Medea cluster. Optionally restrict the result to one node pool."), + func(ctx context.Context, _ *mcp.CallToolRequest, input poolInput) (*mcp.CallToolResult, any, error) { + result, err := queries.ListMachines(ctx, &pb.ListMachinesRequest{Cluster: input.Cluster, Pool: input.Pool}) + return protoResult(result, fmt.Sprintf("Found %d machines in cluster %q.", len(result.GetMachines()), input.Cluster), err) + }) + + mcp.AddTool(server, readTool("list_hosts", "List bare-metal provisioning hosts for one Medea cluster, including allocation, provisioning state, labels, and observed address. Optionally filter by pool."), + func(ctx context.Context, _ *mcp.CallToolRequest, input poolInput) (*mcp.CallToolResult, any, error) { + result, err := queries.ListHosts(ctx, &pb.ListHostsRequest{Cluster: input.Cluster, Pool: input.Pool}) + return protoResult(result, fmt.Sprintf("Found %d provisioning hosts in cluster %q.", len(result.GetHosts()), input.Cluster), err) + }) + + mcp.AddTool(server, readTool("get_rollout", "Get current Kubernetes and per-machine rollout progress for one Medea cluster. Optionally restrict machine rollout progress to one node pool."), + func(ctx context.Context, _ *mcp.CallToolRequest, input poolInput) (*mcp.CallToolResult, any, error) { + result, err := queries.GetRollout(ctx, &pb.GetRolloutRequest{Cluster: input.Cluster, Pool: input.Pool}) + return protoResult(result, fmt.Sprintf("Retrieved rollout state for cluster %q (%d machine records).", input.Cluster, len(result.GetMachineRollouts())), err) + }) + + mcp.AddTool(server, readTool("list_rollouts", "List persisted rollout jobs for one Medea cluster, including kind, target, planned machines, state, creator, and revision."), + func(ctx context.Context, _ *mcp.CallToolRequest, input clusterInput) (*mcp.CallToolResult, any, error) { + result, err := queries.ListRollouts(ctx, &pb.ListRolloutsRequest{Cluster: input.Cluster}) + return protoResult(result, fmt.Sprintf("Found %d rollout jobs for cluster %q.", len(result.GetRollouts()), input.Cluster), err) + }) +} + +func readTool(name, description string) *mcp.Tool { + closedWorld := false + return &mcp.Tool{ + Name: name, Description: description, + Annotations: &mcp.ToolAnnotations{ + ReadOnlyHint: true, IdempotentHint: true, OpenWorldHint: &closedWorld, + }, + } +} + +func protoResult(message proto.Message, summary string, err error) (*mcp.CallToolResult, any, error) { + if err != nil { + return errorResult(err), nil, nil + } + encoded, err := (protojson.MarshalOptions{UseProtoNames: true}).Marshal(message) + if err != nil { + return errorResult(err), nil, nil + } + var structured map[string]any + if err := json.Unmarshal(encoded, &structured); err != nil { + return errorResult(err), nil, nil + } + return &mcp.CallToolResult{ + StructuredContent: structured, + Content: []mcp.Content{&mcp.TextContent{Text: summary + " Use structuredContent for the complete result."}}, + }, nil, nil +}