diff --git a/.gitignore b/.gitignore index 55551b1..9965fb8 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,14 @@ !/README.md !/LICENSE !/docs +!/cmd +!/internal +!/go.mod +!/go.sum +!/Makefile +!/config.example.yaml + +# Build artifacts (belt-and-suspenders; /* already ignores these) +bin/ +*.test + diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..787a075 --- /dev/null +++ b/Makefile @@ -0,0 +1,36 @@ +SHELL := /bin/bash +BINARY := nexus +PKG := gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus +VERSION ?= $(shell git describe --tags --always --dirty 2>/dev/null || echo 0.1.0-dev) +COMMIT ?= $(shell git rev-parse --short HEAD 2>/dev/null || echo unknown) +DATE ?= $(shell date -u +%Y-%m-%dT%H:%M:%SZ) +LDFLAGS := -s -w \ + -X '$(PKG)/internal/version.Version=$(VERSION)' \ + -X '$(PKG)/internal/version.Commit=$(COMMIT)' \ + -X '$(PKG)/internal/version.Date=$(DATE)' + +.PHONY: all build test vet fmt tidy run clean + +all: vet test build + +build: ## Build the nexus binary into ./bin + @mkdir -p bin + go build -ldflags "$(LDFLAGS)" -o bin/$(BINARY) ./cmd/nexus + +test: ## Run all tests + go test ./... + +vet: ## Static analysis + go vet ./... + +fmt: ## Format code + gofmt -s -w . + +tidy: ## Tidy go.mod + go mod tidy + +run: build ## Build and run the gateway with the example config + ./bin/$(BINARY) serve -config config.example.yaml + +clean: + rm -rf bin diff --git a/cmd/nexus/main.go b/cmd/nexus/main.go new file mode 100644 index 0000000..fd1460f --- /dev/null +++ b/cmd/nexus/main.go @@ -0,0 +1,109 @@ +// Command nexus is the MCP Nexus control plane binary. +// +// Subcommands: +// +// nexus serve Run the aggregating MCP gateway (the control plane). +// nexus connect Bridge a local stdio agent to a Nexus gateway endpoint. +// nexus demo-mcp Run the built-in demo MCP server over stdio. +// nexus version Print version information. +package main + +import ( + "context" + "flag" + "fmt" + "log/slog" + "os" + "os/signal" + "syscall" + + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/app" + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/bridge" + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/config" + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/demomcp" + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/version" +) + +func main() { + if len(os.Args) < 2 { + usage() + os.Exit(2) + } + cmd := os.Args[1] + args := os.Args[2:] + + var err error + switch cmd { + case "serve": + err = runServe(args) + case "connect": + err = runConnect(args) + case "demo-mcp": + err = demomcp.Run(os.Stdin, os.Stdout) + case "version", "--version", "-v": + fmt.Println("mcp-nexus", version.String()) + case "help", "-h", "--help": + usage() + default: + fmt.Fprintf(os.Stderr, "unknown command %q\n\n", cmd) + usage() + os.Exit(2) + } + if err != nil { + fmt.Fprintln(os.Stderr, "error:", err) + os.Exit(1) + } +} + +func runServe(args []string) error { + fs := flag.NewFlagSet("serve", flag.ExitOnError) + cfgPath := fs.String("config", "config.yaml", "path to config file") + logFormat := fs.String("log", "text", "log format: text|json") + _ = fs.Parse(args) + + log := newLogger(*logFormat) + + cfg, err := config.Load(*cfgPath) + if err != nil { + return err + } + if cfg.Gateway.Version == "" { + cfg.Gateway.Version = version.Version + } + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + return app.Serve(ctx, cfg, log) +} + +func runConnect(args []string) error { + fs := flag.NewFlagSet("connect", flag.ExitOnError) + url := fs.String("url", "http://localhost:8080/mcp", "Nexus gateway MCP endpoint") + _ = fs.Parse(args) + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + return bridge.Run(ctx, *url, os.Stdin, os.Stdout) +} + +func newLogger(format string) *slog.Logger { + // Logs go to stderr so stdio-transport subcommands keep stdout clean. + var h slog.Handler + if format == "json" { + h = slog.NewJSONHandler(os.Stderr, nil) + } else { + h = slog.NewTextHandler(os.Stderr, nil) + } + return slog.New(h) +} + +func usage() { + fmt.Fprint(os.Stderr, `mcp-nexus — the control plane for MCP + +Usage: + nexus serve [-config config.yaml] [-log text|json] Run the aggregating MCP gateway + nexus connect [-url http://localhost:8080/mcp] Bridge a local stdio agent to a gateway + nexus demo-mcp Run the built-in demo MCP server (stdio) + nexus version Print version +`) +} diff --git a/config.example.yaml b/config.example.yaml new file mode 100644 index 0000000..a157eb3 --- /dev/null +++ b/config.example.yaml @@ -0,0 +1,37 @@ +# MCP Nexus — example configuration (Phase 1). +# +# In Phase 1, upstream MCP servers are declared statically here. Later phases +# populate them automatically via Discovery + Recipes + the Installer. +# +# Environment variables in this file are expanded (e.g. ${HOME}, ${GH_TOKEN}). + +server: + # Address the aggregated MCP endpoint + ops routes listen on. + address: ":8080" + +gateway: + name: "MCP Nexus" + # Path the single aggregated MCP endpoint is served on. + mcp_path: "/mcp" + +upstreams: + # Zero-dependency demo server built into the nexus binary. Great for a first + # end-to-end test: exposes demo.echo, demo.add, demo.now. + - namespace: demo + transport: stdio + command: nexus + args: ["demo-mcp"] + + # Example: the official filesystem MCP server via npx (requires Node). + # Uncomment to expose fs.* tools scoped to a directory. + # - namespace: fs + # transport: stdio + # command: npx + # args: ["-y", "@modelcontextprotocol/server-filesystem", "${HOME}"] + + # Example: a remote MCP server over Streamable HTTP. + # - namespace: remote + # transport: http + # endpoint: "http://127.0.0.1:9000/mcp" + # headers: + # Authorization: "Bearer ${REMOTE_MCP_TOKEN}" diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..4f47a80 --- /dev/null +++ b/go.mod @@ -0,0 +1,5 @@ +module gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus + +go 1.26.4 + +require gopkg.in/yaml.v3 v3.0.1 diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..a62c313 --- /dev/null +++ b/go.sum @@ -0,0 +1,4 @@ +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/app/build.go b/internal/app/build.go new file mode 100644 index 0000000..f39d303 --- /dev/null +++ b/internal/app/build.go @@ -0,0 +1,64 @@ +// Package app wires configuration into the running control/data plane. In +// Phase 1 it builds the Router from static upstream config; later phases feed +// instances from discovery + the reconciler instead. +package app + +import ( + "fmt" + "log/slog" + "os" + + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/config" + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/domain" + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/mcp" + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/router" + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/version" +) + +// nexusInfo identifies Nexus to upstream MCP servers. +func nexusInfo() mcp.Implementation { + return mcp.Implementation{Name: "mcp-nexus", Version: version.Version} +} + +// BuildRouter constructs a Router with an upstream client per config entry. +// Transports are created but not connected; call ConnectAll afterwards. +func BuildRouter(cfg *config.Config, log *slog.Logger) (*router.Router, error) { + r := router.New(nexusInfo(), log) + for _, u := range cfg.Upstreams { + transport, err := buildTransport(u) + if err != nil { + return nil, fmt.Errorf("upstream %q: %w", u.Namespace, err) + } + client := mcp.NewClient(transport, nexusInfo()) + r.Add(u.Namespace, client) + } + return r, nil +} + +func buildTransport(u config.UpstreamConfig) (mcp.Transport, error) { + switch domain.TransportKind(u.Transport) { + case domain.TransportStdio: + return mcp.NewStdioTransport(mcp.StdioConfig{ + Command: u.Command, + Args: u.Args, + Env: u.Env, + Dir: u.Dir, + Stderr: prefixWriter{ns: u.Namespace}, + }) + case domain.TransportHTTP: + return mcp.NewHTTPTransport(mcp.HTTPConfig{ + Endpoint: u.Endpoint, + Headers: u.Headers, + }) + default: + return nil, fmt.Errorf("unknown transport %q", u.Transport) + } +} + +// prefixWriter tags upstream stderr lines with their namespace on our stderr. +type prefixWriter struct{ ns string } + +func (p prefixWriter) Write(b []byte) (int, error) { + fmt.Fprintf(os.Stderr, "[upstream:%s] %s", p.ns, b) + return len(b), nil +} diff --git a/internal/app/serve.go b/internal/app/serve.go new file mode 100644 index 0000000..df0d04c --- /dev/null +++ b/internal/app/serve.go @@ -0,0 +1,69 @@ +package app + +import ( + "context" + "errors" + "log/slog" + "net/http" + "time" + + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/config" + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/gateway" +) + +// Serve builds the router from config, connects upstreams, and runs the HTTP +// gateway until ctx is cancelled, then shuts down gracefully. +func Serve(ctx context.Context, cfg *config.Config, log *slog.Logger) error { + r, err := BuildRouter(cfg, log) + if err != nil { + return err + } + defer r.Close() + + log.Info("connecting upstreams", "count", len(cfg.Upstreams)) + connectCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + r.ConnectAll(connectCtx) + cancel() + for _, s := range r.Statuses() { + if s.Ready { + log.Info("upstream ready", "namespace", s.Namespace, "server", s.Server) + } else { + log.Warn("upstream unavailable", "namespace", s.Namespace, "error", s.Error) + } + } + + gw := gateway.New(gateway.Options{ + Name: cfg.Gateway.Name, + Version: cfg.Gateway.Version, + MCPPath: cfg.Gateway.MCPPath, + Router: r, + Logger: log, + }) + + srv := &http.Server{ + Addr: cfg.Server.Address, + Handler: gw.Handler(), + ReadHeaderTimeout: 10 * time.Second, + } + + errCh := make(chan error, 1) + go func() { + log.Info("gateway listening", + "address", cfg.Server.Address, + "mcp_endpoint", cfg.Gateway.MCPPath, + "tools", len(r.ListTools())) + if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + errCh <- err + } + }() + + select { + case <-ctx.Done(): + log.Info("shutting down") + shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + return srv.Shutdown(shutdownCtx) + case err := <-errCh: + return err + } +} diff --git a/internal/bridge/bridge.go b/internal/bridge/bridge.go new file mode 100644 index 0000000..02edc3a --- /dev/null +++ b/internal/bridge/bridge.go @@ -0,0 +1,78 @@ +// Package bridge implements `nexus connect`: a stdio<->HTTP shim so local +// agents that only speak the MCP stdio transport (Claude Desktop, Cursor, …) +// can attach to the Nexus Gateway's Streamable HTTP endpoint. Each line of +// JSON-RPC on stdin is POSTed to the gateway; the response is written to stdout. +package bridge + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "time" +) + +// Run bridges stdin/stdout to the MCP endpoint at url until stdin closes. +func Run(ctx context.Context, url string, in io.Reader, out io.Writer) error { + client := &http.Client{Timeout: 5 * time.Minute} + scanner := bufio.NewScanner(in) + scanner.Buffer(make([]byte, 0, 64*1024), 16<<20) + enc := json.NewEncoder(out) + + for scanner.Scan() { + line := bytes.TrimSpace(scanner.Bytes()) + if len(line) == 0 { + continue + } + // Preserve the raw frame; forward verbatim. + frame := make([]byte, len(line)) + copy(frame, line) + + respBody, isNotification, err := forward(ctx, client, url, frame) + if err != nil { + return fmt.Errorf("bridge forward: %w", err) + } + if isNotification || len(respBody) == 0 { + continue // notifications produce no stdout frame + } + var msg json.RawMessage = respBody + if err := enc.Encode(msg); err != nil { + return err + } + } + return scanner.Err() +} + +func forward(ctx context.Context, client *http.Client, url string, frame []byte) (body []byte, notification bool, err error) { + // Detect notification (no id) to avoid emitting an empty stdout frame. + var probe struct { + ID json.RawMessage `json:"id"` + Method string `json:"method"` + } + _ = json.Unmarshal(frame, &probe) + notification = probe.Method != "" && len(probe.ID) == 0 + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(frame)) + if err != nil { + return nil, notification, err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json, text/event-stream") + + resp, err := client.Do(req) + if err != nil { + return nil, notification, err + } + defer resp.Body.Close() + b, err := io.ReadAll(io.LimitReader(resp.Body, 32<<20)) + if err != nil { + return nil, notification, err + } + if resp.StatusCode == http.StatusAccepted { + return nil, true, nil + } + return bytes.TrimSpace(b), notification, nil +} diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..5cb1eeb --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,116 @@ +// Package config loads MCP Nexus's declarative YAML configuration. Per +// ARCHITECTURE tenet #2, configuration is the primary way to express desired +// state; in later phases discovery/recipes will also feed it. +package config + +import ( + "fmt" + "os" + "strings" + + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/domain" + "gopkg.in/yaml.v3" +) + +// Config is the top-level configuration document. +type Config struct { + Server ServerConfig `yaml:"server"` + Gateway GatewayConfig `yaml:"gateway"` + Upstreams []UpstreamConfig `yaml:"upstreams"` +} + +// ServerConfig configures the HTTP listener. +type ServerConfig struct { + Address string `yaml:"address"` +} + +// GatewayConfig configures the agent-facing MCP server identity. +type GatewayConfig struct { + Name string `yaml:"name"` + Version string `yaml:"version"` + // MCPPath is the HTTP path the aggregated MCP endpoint is served on. + MCPPath string `yaml:"mcp_path"` +} + +// UpstreamConfig statically declares an upstream MCP server (Phase 1). Later +// phases populate instances via discovery + installer instead. +type UpstreamConfig struct { + Namespace string `yaml:"namespace"` + Transport string `yaml:"transport"` // "stdio" | "http" + Command string `yaml:"command,omitempty"` + Args []string `yaml:"args,omitempty"` + Env []string `yaml:"env,omitempty"` + Dir string `yaml:"dir,omitempty"` + Endpoint string `yaml:"endpoint,omitempty"` + + Headers map[string]string `yaml:"headers,omitempty"` +} + +// Default returns a config populated with sensible defaults. +func Default() *Config { + return &Config{ + Server: ServerConfig{Address: ":8080"}, + Gateway: GatewayConfig{Name: "MCP Nexus", Version: "0.1.0-dev", MCPPath: "/mcp"}, + } +} + +// Load reads, expands environment variables in, and validates a config file. +func Load(path string) (*Config, error) { + raw, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("read config: %w", err) + } + expanded := os.ExpandEnv(string(raw)) + + cfg := Default() + if err := yaml.Unmarshal([]byte(expanded), cfg); err != nil { + return nil, fmt.Errorf("parse config: %w", err) + } + cfg.applyDefaults() + if err := cfg.Validate(); err != nil { + return nil, err + } + return cfg, nil +} + +func (c *Config) applyDefaults() { + if c.Server.Address == "" { + c.Server.Address = ":8080" + } + if c.Gateway.Name == "" { + c.Gateway.Name = "MCP Nexus" + } + if c.Gateway.MCPPath == "" { + c.Gateway.MCPPath = "/mcp" + } + if !strings.HasPrefix(c.Gateway.MCPPath, "/") { + c.Gateway.MCPPath = "/" + c.Gateway.MCPPath + } +} + +// Validate checks the config for internal consistency. +func (c *Config) Validate() error { + seen := map[string]bool{} + for i, u := range c.Upstreams { + if u.Namespace == "" { + return fmt.Errorf("upstreams[%d]: namespace is required", i) + } + if seen[u.Namespace] { + return fmt.Errorf("upstreams[%d]: duplicate namespace %q", i, u.Namespace) + } + seen[u.Namespace] = true + switch domain.TransportKind(u.Transport) { + case domain.TransportStdio: + if u.Command == "" { + return fmt.Errorf("upstreams[%d] (%s): stdio transport requires command", i, u.Namespace) + } + case domain.TransportHTTP: + if u.Endpoint == "" { + return fmt.Errorf("upstreams[%d] (%s): http transport requires endpoint", i, u.Namespace) + } + default: + return fmt.Errorf("upstreams[%d] (%s): unknown transport %q (want stdio|http)", i, u.Namespace, u.Transport) + } + } + return nil +} diff --git a/internal/demomcp/server.go b/internal/demomcp/server.go new file mode 100644 index 0000000..d155ddd --- /dev/null +++ b/internal/demomcp/server.go @@ -0,0 +1,136 @@ +// Package demomcp is a tiny, dependency-free MCP server that speaks the stdio +// transport. It lets you exercise the Nexus Gateway end-to-end with no external +// MCP servers installed, and doubles as a hermetic test fixture. +package demomcp + +import ( + "bufio" + "encoding/json" + "fmt" + "io" + "time" + + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/mcp" +) + +// ServerInfo is the demo server's advertised identity. +var ServerInfo = mcp.Implementation{Name: "nexus-demo-mcp", Version: "0.1.0"} + +// nowFunc is overridable in tests. +var nowFunc = time.Now + +// Run serves MCP over the given reader/writer (typically os.Stdin/os.Stdout) +// until the input is closed. +func Run(in io.Reader, out io.Writer) error { + scanner := bufio.NewScanner(in) + scanner.Buffer(make([]byte, 0, 64*1024), 8<<20) + enc := json.NewEncoder(out) + + for scanner.Scan() { + line := scanner.Bytes() + if len(line) == 0 { + continue + } + var msg mcp.Message + if err := json.Unmarshal(line, &msg); err != nil { + continue + } + if msg.IsNotification() { + continue // e.g. notifications/initialized — nothing to reply + } + resp := dispatch(&msg) + if resp != nil { + if err := enc.Encode(resp); err != nil { + return err + } + } + } + return scanner.Err() +} + +func dispatch(msg *mcp.Message) *mcp.Message { + switch msg.Method { + case mcp.MethodInitialize: + res := mcp.InitializeResult{ + ProtocolVersion: mcp.ProtocolVersion, + Capabilities: mcp.Capabilities{Tools: &mcp.ToolsCapability{}}, + ServerInfo: ServerInfo, + } + m, _ := mcp.NewResult(msg.ID, res) + return m + case mcp.MethodPing: + m, _ := mcp.NewResult(msg.ID, struct{}{}) + return m + case mcp.MethodToolsList: + m, _ := mcp.NewResult(msg.ID, mcp.ListToolsResult{Tools: toolDefs()}) + return m + case mcp.MethodToolsCall: + return callTool(msg) + default: + return mcp.NewError(msg.ID, mcp.CodeMethodNotFound, "unknown method: "+msg.Method) + } +} + +func toolDefs() []mcp.ToolDefinition { + return []mcp.ToolDefinition{ + { + Name: "echo", + Description: "Echo back the provided message.", + InputSchema: map[string]any{ + "type": "object", + "properties": map[string]any{"message": map[string]any{"type": "string"}}, + "required": []any{"message"}, + }, + }, + { + Name: "add", + Description: "Add two numbers and return the sum.", + InputSchema: map[string]any{ + "type": "object", + "properties": map[string]any{ + "a": map[string]any{"type": "number"}, + "b": map[string]any{"type": "number"}, + }, + "required": []any{"a", "b"}, + }, + }, + { + Name: "now", + Description: "Return the current server time in RFC3339.", + InputSchema: map[string]any{"type": "object", "properties": map[string]any{}}, + }, + } +} + +func callTool(msg *mcp.Message) *mcp.Message { + var params mcp.CallToolParams + if err := msg.UnmarshalParams(¶ms); err != nil { + return mcp.NewError(msg.ID, mcp.CodeInvalidParams, err.Error()) + } + switch params.Name { + case "echo": + var args struct { + Message string `json:"message"` + } + _ = json.Unmarshal(params.Arguments, &args) + m, _ := mcp.NewResult(msg.ID, mcp.TextResult(args.Message, false)) + return m + case "add": + var args struct { + A float64 `json:"a"` + B float64 `json:"b"` + } + if err := json.Unmarshal(params.Arguments, &args); err != nil { + m, _ := mcp.NewResult(msg.ID, mcp.TextResult("invalid arguments: "+err.Error(), true)) + return m + } + m, _ := mcp.NewResult(msg.ID, mcp.TextResult(fmt.Sprintf("%g", args.A+args.B), false)) + return m + case "now": + m, _ := mcp.NewResult(msg.ID, mcp.TextResult(nowFunc().UTC().Format(time.RFC3339), false)) + return m + default: + m, _ := mcp.NewResult(msg.ID, mcp.TextResult("unknown tool: "+params.Name, true)) + return m + } +} diff --git a/internal/domain/domain.go b/internal/domain/domain.go new file mode 100644 index 0000000..67eaa88 --- /dev/null +++ b/internal/domain/domain.go @@ -0,0 +1,153 @@ +// Package domain defines the core vocabulary shared across every MCP Nexus +// module, as described in docs/ARCHITECTURE.md §7. These types are the shared +// language of both the data plane (Gateway/Router) and the control plane +// (Discovery/Registry/Installer/…). +package domain + +import "time" + +// Plane distinguishes the two planes of the system (ARCHITECTURE §5). +type Plane string + +const ( + PlaneData Plane = "data" + PlaneControl Plane = "control" +) + +// InstanceState is the lifecycle state of a managed MCP server (see the Health +// Monitoring module doc for the full state machine). +type InstanceState string + +const ( + StateUnknown InstanceState = "unknown" + StatePending InstanceState = "pending" + StateStarting InstanceState = "starting" + StateRunning InstanceState = "running" + StateUnhealthy InstanceState = "unhealthy" + StateOffline InstanceState = "offline" + StateUpdating InstanceState = "updating" + StateRestarting InstanceState = "restarting" + StateQuarantine InstanceState = "quarantined" +) + +// Health is a coarse health verdict for a resource or instance. +type Health string + +const ( + HealthUnknown Health = "unknown" + HealthHealthy Health = "healthy" + HealthDegraded Health = "degraded" + HealthUnhealthy Health = "unhealthy" +) + +// TransportKind identifies how Nexus talks MCP to an upstream server. +type TransportKind string + +const ( + // TransportStdio spawns the server as a subprocess and speaks + // newline-delimited JSON-RPC over stdin/stdout. + TransportStdio TransportKind = "stdio" + // TransportHTTP connects to a server over MCP Streamable HTTP. + TransportHTTP TransportKind = "http" +) + +// DiscoveredResource is a service found on the network by the Discovery Engine. +type DiscoveredResource struct { + UUID string `json:"uuid"` + Type string `json:"type"` + Version string `json:"version,omitempty"` + IP string `json:"ip,omitempty"` + Hostname string `json:"hostname,omitempty"` + Ports []int `json:"ports,omitempty"` + Capabilities []string `json:"capabilities,omitempty"` + Health Health `json:"health"` + Confidence float64 `json:"confidence"` + Source string `json:"source,omitempty"` + Attributes map[string]string `json:"attributes,omitempty"` + FirstSeen time.Time `json:"first_seen"` + LastSeen time.Time `json:"last_seen"` +} + +// PackageSource identifies where an MCP package artifact comes from. +type PackageSource string + +const ( + SourceGitHub PackageSource = "github" + SourceOCI PackageSource = "oci" + SourceDockerHub PackageSource = "dockerhub" + SourceLocal PackageSource = "local" +) + +// Package is a resolvable MCP server artifact from the Registry. +type Package struct { + Name string `json:"name"` + Description string `json:"description,omitempty"` + Source PackageSource `json:"source"` + Ref string `json:"ref"` // image ref or repo + Versions []string `json:"versions,omitempty"` + ConfigSchema string `json:"config_schema,omitempty"` + Signature string `json:"signature,omitempty"` +} + +// Recipe is a declarative match+install+config rule (Smart Recipes module). +type Recipe struct { + Name string `yaml:"name" json:"name"` + Match RecipeMatch `yaml:"match" json:"match"` + Install RecipeInstall `yaml:"install" json:"install"` + Config map[string]string `yaml:"config" json:"config"` +} + +// RecipeMatch is the fingerprint predicate for a recipe. +type RecipeMatch struct { + Type string `yaml:"type,omitempty" json:"type,omitempty"` + Ports []int `yaml:"ports,omitempty" json:"ports,omitempty"` + HTTPTitle []string `yaml:"http_title,omitempty" json:"http_title,omitempty"` +} + +// RecipeInstall describes what to install when a recipe matches. +type RecipeInstall struct { + DockerImage string `yaml:"docker_image,omitempty" json:"docker_image,omitempty"` + Package string `yaml:"package,omitempty" json:"package,omitempty"` +} + +// MCPInstance is a managed, running MCP server — the unit the reconciler drives. +type MCPInstance struct { + ID string `json:"id"` + Namespace string `json:"namespace"` + Package string `json:"package,omitempty"` + Version string `json:"version,omitempty"` + Transport TransportKind `json:"transport"` + State InstanceState `json:"state"` + Health Health `json:"health"` + BoundUUID string `json:"bound_resource_uuid,omitempty"` + ContainerRef string `json:"container_ref,omitempty"` + Config map[string]string `json:"config,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +// Tool is an MCP tool exposed by an instance, namespaced at the Gateway as +// "{namespace}.{name}". +type Tool struct { + Namespace string `json:"namespace"` + Name string `json:"name"` // upstream (un-namespaced) name + QualifiedID string `json:"qualified_id"` // "{namespace}.{name}" + Title string `json:"title,omitempty"` + Description string `json:"description,omitempty"` + // InputSchema is the raw JSON schema object as provided by the upstream. + InputSchema map[string]any `json:"input_schema,omitempty"` +} + +// AgentProfile binds an agent identity to a set of allowed roles. +type AgentProfile struct { + ID string `json:"id"` + Identity string `json:"identity"` + AllowedRoles []string `json:"allowed_roles,omitempty"` +} + +// Role is an RBAC grant of namespaces/tools to principals. +type Role struct { + Name string `json:"name"` + AllowedNamespaces []string `json:"allowed_namespaces,omitempty"` + AllowedTools []string `json:"allowed_tools,omitempty"` +} diff --git a/internal/e2e/e2e_test.go b/internal/e2e/e2e_test.go new file mode 100644 index 0000000..71a4058 --- /dev/null +++ b/internal/e2e/e2e_test.go @@ -0,0 +1,165 @@ +// Package e2e exercises the whole Phase 1 data plane end-to-end: an MCP client +// connected (over in-process pipes) to the built-in demo MCP server, aggregated +// by the Router, and reached through the Gateway's HTTP MCP endpoint. +package e2e + +import ( + "bytes" + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" + + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/demomcp" + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/gateway" + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/mcp" + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/router" +) + +// newDemoClient wires an mcp.Client to a demo server over two pipes, mirroring +// how a real subprocess stdio transport behaves (server closes its output on +// exit so the client's read loop can unblock). +func newDemoClient(t *testing.T) *mcp.Client { + t.Helper() + toServerR, toServerW := io.Pipe() // client -> server + toClientR, toClientW := io.Pipe() // server -> client + go func() { + _ = demomcp.Run(toServerR, toClientW) + _ = toClientW.Close() + }() + transport := mcp.NewStreamTransport(toClientR, toServerW) + return mcp.NewClient(transport, mcp.Implementation{Name: "test", Version: "0"}) +} + +func setup(t *testing.T) (*httptest.Server, *router.Router) { + t.Helper() + r := router.New(mcp.Implementation{Name: "mcp-nexus", Version: "test"}, nil) + r.Add("demo", newDemoClient(t)) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + r.ConnectAll(ctx) + + statuses := r.Statuses() + if len(statuses) != 1 || !statuses[0].Ready { + t.Fatalf("upstream not ready: %+v", statuses) + } + + gw := gateway.New(gateway.Options{Name: "MCP Nexus", Version: "test", MCPPath: "/mcp", Router: r}) + srv := httptest.NewServer(gw.Handler()) + t.Cleanup(func() { srv.Close(); r.Close() }) + return srv, r +} + +// rpc posts a JSON-RPC request to the gateway MCP endpoint and returns the +// decoded response message. +func rpc(t *testing.T, url, method string, params any) *mcp.Message { + t.Helper() + req, err := mcp.NewRequest(json.RawMessage(`1`), method, params) + if err != nil { + t.Fatal(err) + } + body, _ := json.Marshal(req) + resp, err := http.Post(url+"/mcp", "application/json", bytes.NewReader(body)) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + b, _ := io.ReadAll(resp.Body) + var msg mcp.Message + if err := json.Unmarshal(b, &msg); err != nil { + t.Fatalf("decode response: %v (body=%s)", err, b) + } + return &msg +} + +func TestGatewayInitialize(t *testing.T) { + srv, _ := setup(t) + msg := rpc(t, srv.URL, mcp.MethodInitialize, mcp.InitializeParams{ + ProtocolVersion: mcp.ProtocolVersion, + ClientInfo: mcp.Implementation{Name: "agent", Version: "1"}, + }) + if msg.Error != nil { + t.Fatalf("initialize error: %v", msg.Error) + } + var res mcp.InitializeResult + if err := msg.UnmarshalResult(&res); err != nil { + t.Fatal(err) + } + if res.ServerInfo.Name != "MCP Nexus" { + t.Fatalf("serverInfo.name = %q, want MCP Nexus", res.ServerInfo.Name) + } + if res.Capabilities.Tools == nil { + t.Fatal("expected tools capability") + } +} + +func TestGatewayToolsListAggregatesAndNamespaces(t *testing.T) { + srv, _ := setup(t) + msg := rpc(t, srv.URL, mcp.MethodToolsList, mcp.ListToolsParams{}) + if msg.Error != nil { + t.Fatalf("tools/list error: %v", msg.Error) + } + var res mcp.ListToolsResult + if err := msg.UnmarshalResult(&res); err != nil { + t.Fatal(err) + } + want := map[string]bool{"demo.echo": false, "demo.add": false, "demo.now": false} + for _, tool := range res.Tools { + if _, ok := want[tool.Name]; ok { + want[tool.Name] = true + } + } + for name, found := range want { + if !found { + t.Errorf("tool %q not found in aggregated list; got %+v", name, res.Tools) + } + } +} + +func TestGatewayToolCallRoutesToUpstream(t *testing.T) { + srv, _ := setup(t) + + // demo.echo + msg := rpc(t, srv.URL, mcp.MethodToolsCall, mcp.CallToolParams{ + Name: "demo.echo", + Arguments: json.RawMessage(`{"message":"hello nexus"}`), + }) + if msg.Error != nil { + t.Fatalf("tools/call echo error: %v", msg.Error) + } + var res mcp.CallToolResult + if err := msg.UnmarshalResult(&res); err != nil { + t.Fatal(err) + } + if len(res.Content) != 1 || res.Content[0].Text != "hello nexus" { + t.Fatalf("echo result = %+v, want text 'hello nexus'", res.Content) + } + + // demo.add + msg = rpc(t, srv.URL, mcp.MethodToolsCall, mcp.CallToolParams{ + Name: "demo.add", + Arguments: json.RawMessage(`{"a":2,"b":40}`), + }) + if msg.Error != nil { + t.Fatalf("tools/call add error: %v", msg.Error) + } + _ = msg.UnmarshalResult(&res) + if res.Content[0].Text != "42" { + t.Fatalf("add result = %q, want 42", res.Content[0].Text) + } +} + +func TestGatewayUnknownToolReturnsError(t *testing.T) { + srv, _ := setup(t) + msg := rpc(t, srv.URL, mcp.MethodToolsCall, mcp.CallToolParams{Name: "nope.missing"}) + if msg.Error == nil { + t.Fatal("expected error for unknown tool") + } + if msg.Error.Code != mcp.CodeMethodNotFound { + t.Fatalf("error code = %d, want %d", msg.Error.Code, mcp.CodeMethodNotFound) + } +} diff --git a/internal/gateway/gateway.go b/internal/gateway/gateway.go new file mode 100644 index 0000000..e0ea5fd --- /dev/null +++ b/internal/gateway/gateway.go @@ -0,0 +1,224 @@ +// Package gateway implements the agent-facing edge: the single MCP endpoint +// every AI agent connects to. It is an MCP *server* to agents (Streamable HTTP +// + JSON-RPC 2.0) and delegates tool aggregation/dispatch to the Router. +package gateway + +import ( + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "sync/atomic" + "time" + + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/mcp" + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/router" +) + +// Gateway serves the aggregated MCP endpoint plus health/metrics. +type Gateway struct { + info mcp.Implementation + mcpPath string + router Router + log *slog.Logger + + metrics gwMetrics +} + +// Router is the subset of *router.Router the gateway depends on (kept small so +// it is trivially mockable in tests). +type Router interface { + ListTools() []mcp.ToolDefinition + CallTool(ctx context.Context, qualified string, args json.RawMessage) (json.RawMessage, *mcp.RPCError, error) + Statuses() []router.Status +} + +type gwMetrics struct { + requests atomic.Int64 + toolCalls atomic.Int64 + toolErrs atomic.Int64 +} + +// Options configures a Gateway. +type Options struct { + Name string + Version string + MCPPath string + Router Router + Logger *slog.Logger +} + +// New creates a Gateway. +func New(opts Options) *Gateway { + log := opts.Logger + if log == nil { + log = slog.Default() + } + path := opts.MCPPath + if path == "" { + path = "/mcp" + } + return &Gateway{ + info: mcp.Implementation{Name: opts.Name, Version: opts.Version}, + mcpPath: path, + router: opts.Router, + log: log, + } +} + +// Handler returns the HTTP handler exposing the MCP endpoint and ops routes. +func (g *Gateway) Handler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc(g.mcpPath, g.handleMCP) + mux.HandleFunc("/healthz", g.handleHealth) + mux.HandleFunc("/metrics", g.handleMetrics) + mux.HandleFunc("/", g.handleRoot) + return mux +} + +func (g *Gateway) handleRoot(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/" { + http.NotFound(w, r) + return + } + writeJSON(w, http.StatusOK, map[string]any{ + "name": g.info.Name, + "version": g.info.Version, + "mcp": g.mcpPath, + "protocol": mcp.ProtocolVersion, + }) +} + +func (g *Gateway) handleHealth(w http.ResponseWriter, _ *http.Request) { + statuses := g.router.Statuses() + ready := 0 + for _, s := range statuses { + if s.Ready { + ready++ + } + } + writeJSON(w, http.StatusOK, map[string]any{ + "status": "ok", + "upstreams_total": len(statuses), + "upstreams_ready": ready, + "tools": len(g.router.ListTools()), + "upstream_statuses": statuses, + }) +} + +// handleMetrics exposes a minimal Prometheus text exposition. A full +// client_golang integration lands with the Metrics module (Phase 4). +func (g *Gateway) handleMetrics(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + fmt.Fprintf(w, "# HELP nexus_gateway_requests_total Total MCP requests received.\n") + fmt.Fprintf(w, "# TYPE nexus_gateway_requests_total counter\n") + fmt.Fprintf(w, "nexus_gateway_requests_total %d\n", g.metrics.requests.Load()) + fmt.Fprintf(w, "# HELP nexus_gateway_tool_calls_total Total tools/call requests.\n") + fmt.Fprintf(w, "# TYPE nexus_gateway_tool_calls_total counter\n") + fmt.Fprintf(w, "nexus_gateway_tool_calls_total %d\n", g.metrics.toolCalls.Load()) + fmt.Fprintf(w, "# HELP nexus_gateway_tool_errors_total Total failed tools/call requests.\n") + fmt.Fprintf(w, "# TYPE nexus_gateway_tool_errors_total counter\n") + fmt.Fprintf(w, "nexus_gateway_tool_errors_total %d\n", g.metrics.toolErrs.Load()) + fmt.Fprintf(w, "# HELP nexus_gateway_tools Registered tools currently exposed.\n") + fmt.Fprintf(w, "# TYPE nexus_gateway_tools gauge\n") + fmt.Fprintf(w, "nexus_gateway_tools %d\n", len(g.router.ListTools())) +} + +// handleMCP is the single MCP endpoint. It accepts a JSON-RPC request or +// notification and responds with application/json. +func (g *Gateway) handleMCP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + w.Header().Set("Allow", http.MethodPost) + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + g.metrics.requests.Add(1) + + body, err := io.ReadAll(io.LimitReader(r.Body, 32<<20)) + if err != nil { + writeRPC(w, mcp.NewError(nil, mcp.CodeParseError, "read body: "+err.Error())) + return + } + var msg mcp.Message + if err := json.Unmarshal(body, &msg); err != nil { + writeRPC(w, mcp.NewError(nil, mcp.CodeParseError, "invalid JSON-RPC: "+err.Error())) + return + } + + // Notifications get 202 with no body. + if msg.IsNotification() { + w.WriteHeader(http.StatusAccepted) + return + } + resp := g.dispatch(r.Context(), &msg) + writeRPC(w, resp) +} + +func (g *Gateway) dispatch(ctx context.Context, msg *mcp.Message) *mcp.Message { + switch msg.Method { + case mcp.MethodInitialize: + res := mcp.InitializeResult{ + ProtocolVersion: mcp.ProtocolVersion, + Capabilities: mcp.Capabilities{Tools: &mcp.ToolsCapability{ListChanged: true}}, + ServerInfo: g.info, + Instructions: "MCP Nexus aggregates every upstream MCP server behind one endpoint. Tools are namespaced as {namespace}.{tool}.", + } + out, _ := mcp.NewResult(msg.ID, res) + return out + + case mcp.MethodPing: + out, _ := mcp.NewResult(msg.ID, struct{}{}) + return out + + case mcp.MethodToolsList: + res := mcp.ListToolsResult{Tools: g.router.ListTools()} + out, _ := mcp.NewResult(msg.ID, res) + return out + + case mcp.MethodToolsCall: + return g.handleToolCall(ctx, msg) + + default: + return mcp.NewError(msg.ID, mcp.CodeMethodNotFound, "unknown method: "+msg.Method) + } +} + +func (g *Gateway) handleToolCall(ctx context.Context, msg *mcp.Message) *mcp.Message { + g.metrics.toolCalls.Add(1) + var params mcp.CallToolParams + if err := msg.UnmarshalParams(¶ms); err != nil { + g.metrics.toolErrs.Add(1) + return mcp.NewError(msg.ID, mcp.CodeInvalidParams, "invalid params: "+err.Error()) + } + if params.Name == "" { + g.metrics.toolErrs.Add(1) + return mcp.NewError(msg.ID, mcp.CodeInvalidParams, "tool name is required") + } + + start := time.Now() + raw, rpcErr, err := g.router.CallTool(ctx, params.Name, params.Arguments) + if err != nil { + g.metrics.toolErrs.Add(1) + g.log.Warn("tool call transport error", "tool", params.Name, "error", err) + return mcp.NewError(msg.ID, mcp.CodeInternalError, err.Error()) + } + if rpcErr != nil { + g.metrics.toolErrs.Add(1) + return &mcp.Message{JSONRPC: mcp.JSONRPCVersion, ID: msg.ID, Error: rpcErr} + } + g.log.Info("tool call", "tool", params.Name, "ms", time.Since(start).Milliseconds()) + // Pass the upstream MCP result through unmodified. + return &mcp.Message{JSONRPC: mcp.JSONRPCVersion, ID: msg.ID, Result: raw} +} + +func writeRPC(w http.ResponseWriter, msg *mcp.Message) { + writeJSON(w, http.StatusOK, msg) +} + +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(v) +} diff --git a/internal/mcp/client.go b/internal/mcp/client.go new file mode 100644 index 0000000..94631ea --- /dev/null +++ b/internal/mcp/client.go @@ -0,0 +1,122 @@ +package mcp + +import ( + "context" + "encoding/json" + "fmt" + "sync" +) + +// Client is an MCP client bound to a single upstream server via a Transport. +// It performs the initialize handshake and exposes tools/list and tools/call. +type Client struct { + transport Transport + info Implementation + + mu sync.RWMutex + initialized bool + serverInfo Implementation + serverCaps Capabilities +} + +// NewClient wraps a transport. clientInfo identifies Nexus to the upstream. +func NewClient(t Transport, clientInfo Implementation) *Client { + return &Client{transport: t, info: clientInfo} +} + +// SetNotificationHandler forwards peer notifications (e.g. list_changed). +func (c *Client) SetNotificationHandler(h func(*Message)) { + c.transport.SetNotificationHandler(h) +} + +// Initialize performs the MCP initialize handshake and sends the initialized +// notification. It is safe to call once per client. +func (c *Client) Initialize(ctx context.Context) error { + params := InitializeParams{ + ProtocolVersion: ProtocolVersion, + Capabilities: Capabilities{}, + ClientInfo: c.info, + } + resp, err := c.transport.Call(ctx, MethodInitialize, params) + if err != nil { + return fmt.Errorf("initialize: %w", err) + } + if resp.Error != nil { + return fmt.Errorf("initialize: %w", resp.Error) + } + var res InitializeResult + if err := resp.UnmarshalResult(&res); err != nil { + return fmt.Errorf("initialize: decode result: %w", err) + } + c.mu.Lock() + c.initialized = true + c.serverInfo = res.ServerInfo + c.serverCaps = res.Capabilities + c.mu.Unlock() + + if err := c.transport.Notify(ctx, MethodInitialized, nil); err != nil { + return fmt.Errorf("initialized notification: %w", err) + } + return nil +} + +// ServerInfo returns the upstream's advertised identity (valid after Initialize). +func (c *Client) ServerInfo() Implementation { + c.mu.RLock() + defer c.mu.RUnlock() + return c.serverInfo +} + +// ListTools returns all tools exposed by the upstream, following pagination. +func (c *Client) ListTools(ctx context.Context) ([]ToolDefinition, error) { + var all []ToolDefinition + cursor := "" + for { + resp, err := c.transport.Call(ctx, MethodToolsList, ListToolsParams{Cursor: cursor}) + if err != nil { + return nil, fmt.Errorf("tools/list: %w", err) + } + if resp.Error != nil { + return nil, fmt.Errorf("tools/list: %w", resp.Error) + } + var res ListToolsResult + if err := resp.UnmarshalResult(&res); err != nil { + return nil, fmt.Errorf("tools/list: decode: %w", err) + } + all = append(all, res.Tools...) + if res.NextCursor == "" { + break + } + cursor = res.NextCursor + } + return all, nil +} + +// CallTool invokes a tool by its upstream (un-namespaced) name. The raw result +// message is returned so the caller can pass MCP content through unmodified; +// callToolResult is a decoded convenience view. +func (c *Client) CallTool(ctx context.Context, name string, arguments json.RawMessage) (json.RawMessage, *RPCError, error) { + resp, err := c.transport.Call(ctx, MethodToolsCall, CallToolParams{Name: name, Arguments: arguments}) + if err != nil { + return nil, nil, fmt.Errorf("tools/call %s: %w", name, err) + } + if resp.Error != nil { + return nil, resp.Error, nil + } + return resp.Result, nil, nil +} + +// Ping issues an MCP ping, used for liveness checks. +func (c *Client) Ping(ctx context.Context) error { + resp, err := c.transport.Call(ctx, MethodPing, struct{}{}) + if err != nil { + return err + } + if resp.Error != nil { + return resp.Error + } + return nil +} + +// Close shuts down the underlying transport. +func (c *Client) Close() error { return c.transport.Close() } diff --git a/internal/mcp/http_transport.go b/internal/mcp/http_transport.go new file mode 100644 index 0000000..6c2b493 --- /dev/null +++ b/internal/mcp/http_transport.go @@ -0,0 +1,173 @@ +package mcp + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "sync/atomic" +) + +// HTTPConfig configures a Streamable HTTP client transport. +type HTTPConfig struct { + Endpoint string + Headers map[string]string + Client *http.Client +} + +// httpTransport talks MCP over Streamable HTTP: each outbound message is an +// HTTP POST whose response is either a single JSON message or an SSE stream. +type httpTransport struct { + endpoint string + headers map[string]string + client *http.Client + + counter int64 + sessionID atomic.Value // string + notify atomic.Pointer[func(*Message)] +} + +// NewHTTPTransport creates a Streamable HTTP transport to endpoint. +func NewHTTPTransport(cfg HTTPConfig) (Transport, error) { + if cfg.Endpoint == "" { + return nil, fmt.Errorf("http transport: empty endpoint") + } + c := cfg.Client + if c == nil { + c = http.DefaultClient + } + return &httpTransport{endpoint: cfg.Endpoint, headers: cfg.Headers, client: c}, nil +} + +func (t *httpTransport) SetNotificationHandler(h func(*Message)) { + if h == nil { + t.notify.Store(nil) + return + } + t.notify.Store(&h) +} + +func (t *httpTransport) post(ctx context.Context, msg *Message) (*http.Response, error) { + body, err := json.Marshal(msg) + if err != nil { + return nil, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, t.endpoint, bytes.NewReader(body)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json, text/event-stream") + for k, v := range t.headers { + req.Header.Set(k, v) + } + if sid, ok := t.sessionID.Load().(string); ok && sid != "" { + req.Header.Set("Mcp-Session-Id", sid) + } + return t.client.Do(req) +} + +func (t *httpTransport) Call(ctx context.Context, method string, params any) (*Message, error) { + id := atomic.AddInt64(&t.counter, 1) + idRaw := json.RawMessage(fmt.Sprintf("%d", id)) + req, err := NewRequest(idRaw, method, params) + if err != nil { + return nil, err + } + resp, err := t.post(ctx, req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + if sid := resp.Header.Get("Mcp-Session-Id"); sid != "" { + t.sessionID.Store(sid) + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + b, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return nil, fmt.Errorf("http %s: status %d: %s", method, resp.StatusCode, strings.TrimSpace(string(b))) + } + + ct := resp.Header.Get("Content-Type") + want := normalizeID(idRaw) + switch { + case strings.HasPrefix(ct, "text/event-stream"): + return t.readSSE(resp.Body, want) + default: + var m Message + if err := json.NewDecoder(resp.Body).Decode(&m); err != nil { + return nil, fmt.Errorf("http %s: decode: %w", method, err) + } + return &m, nil + } +} + +// readSSE consumes an SSE stream, dispatching peer notifications to the handler +// and returning the response whose id matches want. +func (t *httpTransport) readSSE(r io.Reader, want string) (*Message, error) { + scanner := bufio.NewScanner(r) + scanner.Buffer(make([]byte, 0, 64*1024), 16*1024*1024) + var data strings.Builder + flush := func() (*Message, bool, error) { + if data.Len() == 0 { + return nil, false, nil + } + payload := data.String() + data.Reset() + var m Message + if err := json.Unmarshal([]byte(payload), &m); err != nil { + return nil, false, nil // ignore non-JSON events + } + if m.IsResponse() && normalizeID(m.ID) == want { + return &m, true, nil + } + if m.Method != "" { + if hp := t.notify.Load(); hp != nil { + (*hp)(&m) + } + } + return nil, false, nil + } + for scanner.Scan() { + line := scanner.Text() + if line == "" { // event delimiter + if m, ok, _ := flush(); ok { + return m, nil + } + continue + } + if strings.HasPrefix(line, "data:") { + data.WriteString(strings.TrimSpace(line[len("data:"):])) + } + } + if err := scanner.Err(); err != nil { + return nil, err + } + if m, ok, _ := flush(); ok { + return m, nil + } + return nil, fmt.Errorf("sse stream ended without response for id %s", want) +} + +func (t *httpTransport) Notify(ctx context.Context, method string, params any) error { + note, err := NewNotification(method, params) + if err != nil { + return err + } + resp, err := t.post(ctx, note) + if err != nil { + return err + } + defer resp.Body.Close() + io.Copy(io.Discard, io.LimitReader(resp.Body, 1<<20)) + if resp.StatusCode >= 400 { + return fmt.Errorf("http notify %s: status %d", method, resp.StatusCode) + } + return nil +} + +func (t *httpTransport) Close() error { return nil } diff --git a/internal/mcp/jsonrpc.go b/internal/mcp/jsonrpc.go new file mode 100644 index 0000000..fd7de4d --- /dev/null +++ b/internal/mcp/jsonrpc.go @@ -0,0 +1,109 @@ +// Package mcp implements the Model Context Protocol (MCP) over JSON-RPC 2.0, +// providing both a client (Nexus -> upstream MCP servers) and the primitives +// used by the agent-facing MCP server (agents -> Nexus Gateway). +package mcp + +import ( + "encoding/json" + "fmt" +) + +// JSONRPCVersion is the only JSON-RPC version MCP uses. +const JSONRPCVersion = "2.0" + +// Standard JSON-RPC 2.0 error codes. +const ( + CodeParseError = -32700 + CodeInvalidRequest = -32600 + CodeMethodNotFound = -32601 + CodeInvalidParams = -32602 + CodeInternalError = -32603 +) + +// Message is a single JSON-RPC 2.0 frame. It is deliberately permissive so it +// can represent requests, responses, and notifications on the wire; use the +// helpers to interpret it. +type Message struct { + JSONRPC string `json:"jsonrpc"` + ID json.RawMessage `json:"id,omitempty"` + Method string `json:"method,omitempty"` + Params json.RawMessage `json:"params,omitempty"` + Result json.RawMessage `json:"result,omitempty"` + Error *RPCError `json:"error,omitempty"` +} + +// IsRequest reports whether the message is a request (has method and id). +func (m *Message) IsRequest() bool { return m.Method != "" && len(m.ID) > 0 } + +// IsNotification reports whether the message is a notification (method, no id). +func (m *Message) IsNotification() bool { return m.Method != "" && len(m.ID) == 0 } + +// IsResponse reports whether the message is a response (result or error, id). +func (m *Message) IsResponse() bool { return m.Method == "" && len(m.ID) > 0 } + +// RPCError is a JSON-RPC 2.0 error object. +type RPCError struct { + Code int `json:"code"` + Message string `json:"message"` + Data json.RawMessage `json:"data,omitempty"` +} + +func (e *RPCError) Error() string { + return fmt.Sprintf("jsonrpc error %d: %s", e.Code, e.Message) +} + +// NewRequest builds a request message with the given id, method, and params. +func NewRequest(id json.RawMessage, method string, params any) (*Message, error) { + m := &Message{JSONRPC: JSONRPCVersion, ID: id, Method: method} + if params != nil { + b, err := json.Marshal(params) + if err != nil { + return nil, err + } + m.Params = b + } + return m, nil +} + +// NewNotification builds a notification message (no id). +func NewNotification(method string, params any) (*Message, error) { + m := &Message{JSONRPC: JSONRPCVersion, Method: method} + if params != nil { + b, err := json.Marshal(params) + if err != nil { + return nil, err + } + m.Params = b + } + return m, nil +} + +// NewResult builds a successful response for the given request id. +func NewResult(id json.RawMessage, result any) (*Message, error) { + b, err := json.Marshal(result) + if err != nil { + return nil, err + } + return &Message{JSONRPC: JSONRPCVersion, ID: id, Result: b}, nil +} + +// NewError builds an error response for the given request id. +func NewError(id json.RawMessage, code int, msg string) *Message { + return &Message{JSONRPC: JSONRPCVersion, ID: id, Error: &RPCError{Code: code, Message: msg}} +} + +// UnmarshalParams decodes the params of the message into v. +func (m *Message) UnmarshalParams(v any) error { + if len(m.Params) == 0 { + return nil + } + return json.Unmarshal(m.Params, v) +} + +// UnmarshalResult decodes the result of the message into v. +func (m *Message) UnmarshalResult(v any) error { + if len(m.Result) == 0 { + return nil + } + return json.Unmarshal(m.Result, v) +} diff --git a/internal/mcp/protocol.go b/internal/mcp/protocol.go new file mode 100644 index 0000000..4cd72c1 --- /dev/null +++ b/internal/mcp/protocol.go @@ -0,0 +1,106 @@ +package mcp + +import "encoding/json" + +// ProtocolVersion is the MCP protocol revision Nexus advertises. Nexus is +// lenient on the client's requested version and echoes a supported one. +const ProtocolVersion = "2025-06-18" + +// Method names used by MCP. +const ( + MethodInitialize = "initialize" + MethodInitialized = "notifications/initialized" + MethodPing = "ping" + MethodToolsList = "tools/list" + MethodToolsCall = "tools/call" + MethodToolListChged = "notifications/tools/list_changed" +) + +// Implementation identifies a client or server implementation. +type Implementation struct { + Name string `json:"name"` + Version string `json:"version"` +} + +// ToolsCapability describes tool-related server capabilities. +type ToolsCapability struct { + ListChanged bool `json:"listChanged,omitempty"` +} + +// Capabilities is the (subset of) MCP capabilities Nexus cares about. +type Capabilities struct { + Tools *ToolsCapability `json:"tools,omitempty"` + Resources *struct { + ListChanged bool `json:"listChanged,omitempty"` + Subscribe bool `json:"subscribe,omitempty"` + } `json:"resources,omitempty"` + Prompts *struct { + ListChanged bool `json:"listChanged,omitempty"` + } `json:"prompts,omitempty"` +} + +// InitializeParams are the params for the initialize request. +type InitializeParams struct { + ProtocolVersion string `json:"protocolVersion"` + Capabilities Capabilities `json:"capabilities"` + ClientInfo Implementation `json:"clientInfo"` +} + +// InitializeResult is the result of an initialize request. +type InitializeResult struct { + ProtocolVersion string `json:"protocolVersion"` + Capabilities Capabilities `json:"capabilities"` + ServerInfo Implementation `json:"serverInfo"` + Instructions string `json:"instructions,omitempty"` +} + +// ToolDefinition is a tool as described by tools/list. +type ToolDefinition struct { + Name string `json:"name"` + Title string `json:"title,omitempty"` + Description string `json:"description,omitempty"` + InputSchema map[string]any `json:"inputSchema,omitempty"` +} + +// ListToolsParams are the params for tools/list. +type ListToolsParams struct { + Cursor string `json:"cursor,omitempty"` +} + +// ListToolsResult is the result of tools/list. +type ListToolsResult struct { + Tools []ToolDefinition `json:"tools"` + NextCursor string `json:"nextCursor,omitempty"` +} + +// CallToolParams are the params for tools/call. +type CallToolParams struct { + Name string `json:"name"` + Arguments json.RawMessage `json:"arguments,omitempty"` +} + +// ContentBlock is one item of tool result content. Only the common fields are +// modeled; unknown fields survive round-trips because callers pass the raw +// result through where possible. +type ContentBlock struct { + Type string `json:"type"` + Text string `json:"text,omitempty"` + // For image/audio/resource blocks these carry through opaquely. + Data string `json:"data,omitempty"` + MimeType string `json:"mimeType,omitempty"` + Resource json.RawMessage `json:"resource,omitempty"` +} + +// CallToolResult is the result of tools/call. +type CallToolResult struct { + Content []ContentBlock `json:"content"` + IsError bool `json:"isError,omitempty"` +} + +// TextResult is a convenience constructor for a single text-content result. +func TextResult(text string, isError bool) CallToolResult { + return CallToolResult{ + Content: []ContentBlock{{Type: "text", Text: text}}, + IsError: isError, + } +} diff --git a/internal/mcp/stdio_transport.go b/internal/mcp/stdio_transport.go new file mode 100644 index 0000000..679c57a --- /dev/null +++ b/internal/mcp/stdio_transport.go @@ -0,0 +1,241 @@ +package mcp + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "os/exec" + "strconv" + "sync" + "sync/atomic" +) + +// StdioConfig configures a subprocess-backed stdio MCP transport. +type StdioConfig struct { + Command string + Args []string + Env []string // additional environment (KEY=VALUE), appended to os env + Dir string + // Stderr, if non-nil, receives the child process's stderr. + Stderr io.Writer +} + +// streamTransport speaks newline-delimited JSON-RPC over any reader/writer +// pair, correlating responses with requests by id. It backs both the +// subprocess stdio transport and in-process pipe-connected transports. +type streamTransport struct { + w io.WriteCloser + enc *json.Encoder + r io.Reader + + // process control (nil for non-subprocess transports) + proc interface{ Kill() error } + waitFn func() error + + mu sync.Mutex + counter int64 + pending map[string]chan *Message + closed bool + exitErr error + + notify atomic.Pointer[func(*Message)] + done chan struct{} +} + +func newStreamTransport(r io.Reader, w io.WriteCloser) *streamTransport { + t := &streamTransport{ + w: w, + enc: json.NewEncoder(w), + r: r, + pending: make(map[string]chan *Message), + done: make(chan struct{}), + } + go t.readLoop() + return t +} + +// NewStreamTransport creates a transport over an existing reader/writer pair, +// e.g. an in-process pipe. It performs no process management. +func NewStreamTransport(r io.Reader, w io.WriteCloser) Transport { + return newStreamTransport(r, w) +} + +// NewStdioTransport spawns the configured command and serves MCP over its +// stdio. The process is terminated by Close. +func NewStdioTransport(cfg StdioConfig) (Transport, error) { + if cfg.Command == "" { + return nil, errors.New("stdio transport: empty command") + } + cmd := exec.Command(cfg.Command, cfg.Args...) + cmd.Dir = cfg.Dir + if len(cfg.Env) > 0 { + cmd.Env = append(cmd.Environ(), cfg.Env...) + } + cmd.Stderr = cfg.Stderr + + stdin, err := cmd.StdinPipe() + if err != nil { + return nil, fmt.Errorf("stdio transport: stdin pipe: %w", err) + } + stdout, err := cmd.StdoutPipe() + if err != nil { + return nil, fmt.Errorf("stdio transport: stdout pipe: %w", err) + } + if err := cmd.Start(); err != nil { + return nil, fmt.Errorf("stdio transport: start %q: %w", cfg.Command, err) + } + + t := &streamTransport{ + w: stdin, + enc: json.NewEncoder(stdin), + r: stdout, + waitFn: cmd.Wait, + pending: make(map[string]chan *Message), + done: make(chan struct{}), + } + if cmd.Process != nil { + t.proc = cmd.Process + } + go t.readLoop() + return t, nil +} + +func (t *streamTransport) SetNotificationHandler(h func(*Message)) { + if h == nil { + t.notify.Store(nil) + return + } + t.notify.Store(&h) +} + +func (t *streamTransport) readLoop() { + scanner := bufio.NewScanner(t.r) + // MCP messages can be large (tool schemas); allow up to 16 MiB lines. + scanner.Buffer(make([]byte, 0, 64*1024), 16*1024*1024) + for scanner.Scan() { + line := scanner.Bytes() + if len(bytes.TrimSpace(line)) == 0 { + continue + } + var m Message + if err := json.Unmarshal(line, &m); err != nil { + continue // ignore non-JSON noise + } + if m.IsResponse() { + key := normalizeID(m.ID) + t.mu.Lock() + ch := t.pending[key] + delete(t.pending, key) + t.mu.Unlock() + if ch != nil { + ch <- &m + } + continue + } + if m.Method != "" { + if hp := t.notify.Load(); hp != nil { + (*hp)(&m) + } + } + } + // input closed: peer is gone. Fail all in-flight calls. + var exitErr error + if t.waitFn != nil { + exitErr = t.waitFn() + } + t.mu.Lock() + t.closed = true + t.exitErr = exitErr + pending := t.pending + t.pending = map[string]chan *Message{} + t.mu.Unlock() + for _, ch := range pending { + close(ch) + } + close(t.done) +} + +func (t *streamTransport) Call(ctx context.Context, method string, params any) (*Message, error) { + id := atomic.AddInt64(&t.counter, 1) + idRaw := json.RawMessage(strconv.FormatInt(id, 10)) + req, err := NewRequest(idRaw, method, params) + if err != nil { + return nil, err + } + key := normalizeID(idRaw) + ch := make(chan *Message, 1) + + t.mu.Lock() + if t.closed { + t.mu.Unlock() + return nil, t.closedErr() + } + t.pending[key] = ch + err = t.enc.Encode(req) + t.mu.Unlock() + if err != nil { + t.mu.Lock() + delete(t.pending, key) + t.mu.Unlock() + return nil, fmt.Errorf("stdio write %s: %w", method, err) + } + + select { + case <-ctx.Done(): + t.mu.Lock() + delete(t.pending, key) + t.mu.Unlock() + return nil, ctx.Err() + case resp, ok := <-ch: + if !ok { + return nil, t.closedErr() + } + return resp, nil + case <-t.done: + return nil, t.closedErr() + } +} + +func (t *streamTransport) Notify(_ context.Context, method string, params any) error { + note, err := NewNotification(method, params) + if err != nil { + return err + } + t.mu.Lock() + defer t.mu.Unlock() + if t.closed { + return t.closedErr() + } + return t.enc.Encode(note) +} + +func (t *streamTransport) closedErr() error { + if t.exitErr != nil { + return fmt.Errorf("mcp stdio transport closed: %w", t.exitErr) + } + return errors.New("mcp stdio transport closed") +} + +func (t *streamTransport) Close() error { + t.mu.Lock() + closed := t.closed + t.mu.Unlock() + _ = t.w.Close() + if t.proc != nil { + _ = t.proc.Kill() + } + if !closed { + <-t.done + } + return nil +} + +// normalizeID produces a stable map key from a JSON-RPC id, tolerating +// whitespace differences in how peers echo the id. +func normalizeID(raw json.RawMessage) string { + return string(bytes.TrimSpace(raw)) +} diff --git a/internal/mcp/transport.go b/internal/mcp/transport.go new file mode 100644 index 0000000..1c92bdc --- /dev/null +++ b/internal/mcp/transport.go @@ -0,0 +1,19 @@ +package mcp + +import "context" + +// Transport is a bidirectional MCP channel to a single peer. Implementations +// correlate requests with responses internally so callers see simple +// request/response semantics. +// +// Call returns the response Message (which may itself carry a JSON-RPC error in +// Message.Error); the returned error is reserved for transport-level failures +// (process died, connection refused, context cancelled, malformed frame). +type Transport interface { + Call(ctx context.Context, method string, params any) (*Message, error) + Notify(ctx context.Context, method string, params any) error + // SetNotificationHandler registers a handler for peer-initiated + // notifications/requests (e.g. tools/list_changed). May be nil. + SetNotificationHandler(func(*Message)) + Close() error +} diff --git a/internal/router/registry.go b/internal/router/registry.go new file mode 100644 index 0000000..0752110 --- /dev/null +++ b/internal/router/registry.go @@ -0,0 +1,129 @@ +// Package router implements the Dynamic Tool Registry and the MCP Router: +// it aggregates every upstream MCP server, namespaces their tools to avoid +// collisions, and dispatches tools/call to the owning upstream (see the +// Gateway and Dynamic Tool Registry module docs). +package router + +import ( + "sort" + "strings" + "sync" + + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/domain" + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/mcp" +) + +// NamespaceSep separates a namespace from a tool name in a qualified id. +const NamespaceSep = "." + +// Qualify builds the agent-facing tool id "{namespace}.{name}". +func Qualify(namespace, name string) string { + return namespace + NamespaceSep + name +} + +// SplitQualified splits a qualified id into namespace and upstream tool name on +// the first separator. ok is false if there is no separator. +func SplitQualified(qualified string) (namespace, name string, ok bool) { + idx := strings.Index(qualified, NamespaceSep) + if idx <= 0 || idx == len(qualified)-1 { + return "", "", false + } + return qualified[:idx], qualified[idx+1:], true +} + +type indexed struct { + namespace string + name string // upstream (un-namespaced) tool name + def mcp.ToolDefinition +} + +// Registry is the live, namespaced catalog of every tool exposed by every +// healthy upstream. It is safe for concurrent use. +type Registry struct { + mu sync.RWMutex + tools map[string]indexed // qualified id -> entry +} + +// NewRegistry creates an empty registry. +func NewRegistry() *Registry { + return &Registry{tools: make(map[string]indexed)} +} + +// Replace atomically swaps the full set of tools for a namespace. +func (r *Registry) Replace(namespace string, defs []mcp.ToolDefinition) { + r.mu.Lock() + defer r.mu.Unlock() + // drop existing tools for this namespace + for q, e := range r.tools { + if e.namespace == namespace { + delete(r.tools, q) + } + } + for _, d := range defs { + q := Qualify(namespace, d.Name) + r.tools[q] = indexed{namespace: namespace, name: d.Name, def: d} + } +} + +// Remove drops all tools belonging to a namespace (e.g. upstream went away). +func (r *Registry) Remove(namespace string) { + r.mu.Lock() + defer r.mu.Unlock() + for q, e := range r.tools { + if e.namespace == namespace { + delete(r.tools, q) + } + } +} + +// Lookup resolves a qualified id to its owning namespace and upstream tool name. +func (r *Registry) Lookup(qualified string) (namespace, name string, ok bool) { + r.mu.RLock() + defer r.mu.RUnlock() + e, found := r.tools[qualified] + if !found { + return "", "", false + } + return e.namespace, e.name, true +} + +// Definitions returns the namespaced tool definitions to advertise to agents +// via tools/list. Names are the qualified ids. +func (r *Registry) Definitions() []mcp.ToolDefinition { + r.mu.RLock() + defer r.mu.RUnlock() + out := make([]mcp.ToolDefinition, 0, len(r.tools)) + for q, e := range r.tools { + d := e.def + d.Name = q // present the namespaced id to agents + out = append(out, d) + } + sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) + return out +} + +// List returns the catalog as domain.Tool records (for APIs/dashboard). +func (r *Registry) List() []domain.Tool { + r.mu.RLock() + defer r.mu.RUnlock() + out := make([]domain.Tool, 0, len(r.tools)) + for q, e := range r.tools { + out = append(out, domain.Tool{ + Namespace: e.namespace, + Name: e.name, + QualifiedID: q, + Title: e.def.Title, + Description: e.def.Description, + InputSchema: e.def.InputSchema, + }) + } + sort.Slice(out, func(i, j int) bool { return out[i].QualifiedID < out[j].QualifiedID }) + return out +} + +// Len returns the number of registered tools. +func (r *Registry) Len() int { + r.mu.RLock() + defer r.mu.RUnlock() + return len(r.tools) +} diff --git a/internal/router/registry_test.go b/internal/router/registry_test.go new file mode 100644 index 0000000..4b11904 --- /dev/null +++ b/internal/router/registry_test.go @@ -0,0 +1,66 @@ +package router + +import ( + "testing" + + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/mcp" +) + +func TestQualifyAndSplit(t *testing.T) { + q := Qualify("demo", "echo") + if q != "demo.echo" { + t.Fatalf("Qualify = %q, want demo.echo", q) + } + ns, name, ok := SplitQualified("demo.echo") + if !ok || ns != "demo" || name != "echo" { + t.Fatalf("SplitQualified = (%q,%q,%v)", ns, name, ok) + } + // tool name containing a dot: split on first separator only + ns, name, ok = SplitQualified("fs.read.file") + if !ok || ns != "fs" || name != "read.file" { + t.Fatalf("SplitQualified dotted = (%q,%q,%v)", ns, name, ok) + } + if _, _, ok := SplitQualified("nodot"); ok { + t.Fatal("SplitQualified with no separator should fail") + } +} + +func TestRegistryReplaceLookupDefinitions(t *testing.T) { + r := NewRegistry() + r.Replace("demo", []mcp.ToolDefinition{ + {Name: "echo", Description: "echo"}, + {Name: "add", Description: "add"}, + }) + r.Replace("fs", []mcp.ToolDefinition{{Name: "read"}}) + + if r.Len() != 3 { + t.Fatalf("Len = %d, want 3", r.Len()) + } + ns, name, ok := r.Lookup("demo.echo") + if !ok || ns != "demo" || name != "echo" { + t.Fatalf("Lookup demo.echo = (%q,%q,%v)", ns, name, ok) + } + + defs := r.Definitions() + if len(defs) != 3 { + t.Fatalf("Definitions len = %d, want 3", len(defs)) + } + // definitions must expose the namespaced id and be sorted + if defs[0].Name != "demo.add" { + t.Fatalf("first def = %q, want demo.add", defs[0].Name) + } + + // replacing a namespace must not leave stale tools + r.Replace("demo", []mcp.ToolDefinition{{Name: "echo"}}) + if _, _, ok := r.Lookup("demo.add"); ok { + t.Fatal("demo.add should be gone after replace") + } + if r.Len() != 2 { + t.Fatalf("Len after replace = %d, want 2", r.Len()) + } + + r.Remove("fs") + if _, _, ok := r.Lookup("fs.read"); ok { + t.Fatal("fs.read should be gone after Remove") + } +} diff --git a/internal/router/router.go b/internal/router/router.go new file mode 100644 index 0000000..fee3101 --- /dev/null +++ b/internal/router/router.go @@ -0,0 +1,189 @@ +package router + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "sync" + + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/mcp" +) + +// Upstream is a single aggregated MCP server behind the Gateway. +type Upstream struct { + Namespace string + Client *mcp.Client + + mu sync.RWMutex + ready bool + server mcp.Implementation + lastErr error +} + +func (u *Upstream) setReady(server mcp.Implementation) { + u.mu.Lock() + defer u.mu.Unlock() + u.ready = true + u.server = server + u.lastErr = nil +} + +func (u *Upstream) setErr(err error) { + u.mu.Lock() + defer u.mu.Unlock() + u.ready = false + u.lastErr = err +} + +// Status is a snapshot of an upstream's connection state. +type Status struct { + Namespace string `json:"namespace"` + Ready bool `json:"ready"` + Server string `json:"server,omitempty"` + Error string `json:"error,omitempty"` +} + +// Router aggregates upstreams and routes tool calls, backed by the Registry. +type Router struct { + info mcp.Implementation + log *slog.Logger + registry *Registry + + mu sync.RWMutex + ups map[string]*Upstream // by namespace +} + +// New creates a Router. info identifies Nexus to upstreams. +func New(info mcp.Implementation, log *slog.Logger) *Router { + if log == nil { + log = slog.Default() + } + return &Router{ + info: info, + log: log, + registry: NewRegistry(), + ups: make(map[string]*Upstream), + } +} + +// Registry exposes the tool catalog. +func (r *Router) Registry() *Registry { return r.registry } + +// Add registers an upstream client under a namespace. It does not connect. +func (r *Router) Add(namespace string, client *mcp.Client) *Upstream { + u := &Upstream{Namespace: namespace, Client: client} + r.mu.Lock() + r.ups[namespace] = u + r.mu.Unlock() + return u +} + +// ConnectAll initializes every upstream and loads its tools, in parallel. +// Individual failures are logged and leave that upstream not-ready rather than +// failing the whole gateway (the data plane must stay up — ARCHITECTURE §5). +func (r *Router) ConnectAll(ctx context.Context) { + r.mu.RLock() + ups := make([]*Upstream, 0, len(r.ups)) + for _, u := range r.ups { + ups = append(ups, u) + } + r.mu.RUnlock() + + var wg sync.WaitGroup + for _, u := range ups { + wg.Add(1) + go func(u *Upstream) { + defer wg.Done() + if err := r.connect(ctx, u); err != nil { + u.setErr(err) + r.log.Warn("upstream connect failed", "namespace", u.Namespace, "error", err) + } + }(u) + } + wg.Wait() +} + +func (r *Router) connect(ctx context.Context, u *Upstream) error { + if err := u.Client.Initialize(ctx); err != nil { + return err + } + u.setReady(u.Client.ServerInfo()) + // Refresh tools list on upstream-initiated change notifications. + u.Client.SetNotificationHandler(func(m *mcp.Message) { + if m.Method == mcp.MethodToolListChged { + r.log.Info("upstream tools changed", "namespace", u.Namespace) + // Best-effort refresh in the background. + go func() { + if err := r.RefreshTools(context.Background(), u); err != nil { + r.log.Warn("tool refresh failed", "namespace", u.Namespace, "error", err) + } + }() + } + }) + return r.RefreshTools(ctx, u) +} + +// RefreshTools reloads and re-indexes a single upstream's tools. +func (r *Router) RefreshTools(ctx context.Context, u *Upstream) error { + defs, err := u.Client.ListTools(ctx) + if err != nil { + return err + } + r.registry.Replace(u.Namespace, defs) + r.log.Info("indexed upstream tools", "namespace", u.Namespace, "count", len(defs)) + return nil +} + +// ListTools returns the aggregated, namespaced tool definitions for agents. +func (r *Router) ListTools() []mcp.ToolDefinition { + return r.registry.Definitions() +} + +// CallTool routes a namespaced tool call to its owning upstream and returns the +// raw MCP result (passed through unmodified) or a JSON-RPC error. +func (r *Router) CallTool(ctx context.Context, qualified string, args json.RawMessage) (json.RawMessage, *mcp.RPCError, error) { + namespace, name, ok := r.registry.Lookup(qualified) + if !ok { + return nil, &mcp.RPCError{Code: mcp.CodeMethodNotFound, Message: fmt.Sprintf("unknown tool %q", qualified)}, nil + } + r.mu.RLock() + u := r.ups[namespace] + r.mu.RUnlock() + if u == nil { + return nil, &mcp.RPCError{Code: mcp.CodeInternalError, Message: fmt.Sprintf("no upstream for namespace %q", namespace)}, nil + } + u.mu.RLock() + ready := u.ready + u.mu.RUnlock() + if !ready { + return nil, &mcp.RPCError{Code: mcp.CodeInternalError, Message: fmt.Sprintf("upstream %q is not ready", namespace)}, nil + } + return u.Client.CallTool(ctx, name, args) +} + +// Statuses returns a snapshot of every upstream's connection state. +func (r *Router) Statuses() []Status { + r.mu.RLock() + defer r.mu.RUnlock() + out := make([]Status, 0, len(r.ups)) + for _, u := range r.ups { + u.mu.RLock() + s := Status{Namespace: u.Namespace, Ready: u.ready, Server: u.server.Name} + if u.lastErr != nil { + s.Error = u.lastErr.Error() + } + u.mu.RUnlock() + out = append(out, s) + } + return out +} + +// Close shuts down every upstream client. +func (r *Router) Close() { + r.mu.Lock() + defer r.mu.Unlock() + for _, u := range r.ups { + _ = u.Client.Close() + } +} diff --git a/internal/store/store.go b/internal/store/store.go new file mode 100644 index 0000000..358434d --- /dev/null +++ b/internal/store/store.go @@ -0,0 +1,72 @@ +// Package store defines MCP Nexus's persistence interfaces (ARCHITECTURE §8) +// and an in-memory implementation used for Phase 1 and tests. A SQLite-backed +// implementation lands in a later brick behind the same interfaces. +package store + +import ( + "context" + "sort" + "sync" + + "gitea.thetempleofdoom.com/drjones/mcp-gateway-nexus/internal/domain" +) + +// InstanceStore persists MCP instances (the observed/desired state the +// reconciler manages). +type InstanceStore interface { + PutInstance(ctx context.Context, inst domain.MCPInstance) error + GetInstance(ctx context.Context, id string) (domain.MCPInstance, bool, error) + ListInstances(ctx context.Context) ([]domain.MCPInstance, error) + DeleteInstance(ctx context.Context, id string) error +} + +// Store aggregates all persistence surfaces. It will grow as later modules +// (registry, identity, secrets, audit) come online. +type Store interface { + InstanceStore +} + +// Memory is a thread-safe in-memory Store. +type Memory struct { + mu sync.RWMutex + instances map[string]domain.MCPInstance +} + +// NewMemory creates an empty in-memory store. +func NewMemory() *Memory { + return &Memory{instances: make(map[string]domain.MCPInstance)} +} + +func (m *Memory) PutInstance(_ context.Context, inst domain.MCPInstance) error { + m.mu.Lock() + defer m.mu.Unlock() + m.instances[inst.ID] = inst + return nil +} + +func (m *Memory) GetInstance(_ context.Context, id string) (domain.MCPInstance, bool, error) { + m.mu.RLock() + defer m.mu.RUnlock() + inst, ok := m.instances[id] + return inst, ok, nil +} + +func (m *Memory) ListInstances(_ context.Context) ([]domain.MCPInstance, error) { + m.mu.RLock() + defer m.mu.RUnlock() + out := make([]domain.MCPInstance, 0, len(m.instances)) + for _, inst := range m.instances { + out = append(out, inst) + } + sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) + return out, nil +} + +func (m *Memory) DeleteInstance(_ context.Context, id string) error { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.instances, id) + return nil +} + +var _ Store = (*Memory)(nil) diff --git a/internal/version/version.go b/internal/version/version.go new file mode 100644 index 0000000..36ae899 --- /dev/null +++ b/internal/version/version.go @@ -0,0 +1,17 @@ +// Package version holds build-time version information for MCP Nexus. +package version + +// These are overridable at build time via -ldflags. +var ( + // Version is the semantic version of the build. + Version = "0.1.0-dev" + // Commit is the git commit the binary was built from. + Commit = "unknown" + // Date is the build date. + Date = "unknown" +) + +// String returns a human-readable version string. +func String() string { + return Version + " (commit " + Commit + ", built " + Date + ")" +}