From 95ae900a588da314b0862f361e7967af3b06cc54 Mon Sep 17 00:00:00 2001 From: kami Date: Sat, 1 Aug 2026 04:22:52 +0400 Subject: [PATCH] Talk MCP: a client for external tool servers (#251) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit docs/plans/06-mcp-support.md asks for the host direction — Maven connects OUT to MCP servers and consumes what they offer. This is the client half: the protocol, the transports, the connection manager, the config block. Nothing is wired into a turn yet, and nothing here exposes Maven's own capabilities to an outside caller. internal/mcp: - hand-rolled JSON-RPC 2.0 (the wire format is four fields, and the repo vendors its deps, so a library would cost more than it saves); - two transports: a stdio subprocess on this box, and streamable HTTP, which accepts a plain JSON reply or an SSE frame because servers disagree about which they send; - Client: initialize handshake, tools/list, tools/call, resources/list, resources/read. Text content only — everything downstream is a sentence; - Manager: lazy dial, per-server failure that never blocks boot or the other servers, backoff reconnect, Status for a web surface, graceful Close; - the allowlist encoding: a discovered tool becomes the store row "vikunja_list_tasks" with cmd ["mcp","vikunja","list_tasks"], scope "mcp:vikunja". No new column, no migration, and ProposeTool, EnableTool, the act matcher and the confirm turn all keep working untouched. Constraints held, in code rather than in prose: - OFF unless configured, and a server is dark until "enabled": true. - A url server goes through internal/webfetch, so the SSRF guard, the size cap, the redirect cap and the per-host rate limit apply. Reaching loopback needs allow_private on THAT server, and each server gets its own fetcher so one loopback exemption cannot become a hole for a public endpoint. - readOnlyHint decides destructive: no hint means "assume it mutates", which will route the call through the existing confirm turn. Guessing wrong in that direction only costs a question. - The catalogue stays small on purpose — allow_tools, and max_tools=12 per server. The resident model is a 1.7B with a 4096-token context; a tool name it half-remembers is a wrong act. - Only the tool name and the router's arguments are sent. There is no API here through which a note, a fact or the persona block could travel. webfetch grows Post (JSON-RPC cannot be a GET) and surfaces response headers for Mcp-Session-Id. It shares Get's guards exactly: a body buys a caller nothing, a POST to the LAN is refused for the same reason a GET is. Verified against the real Vikunja MCP server on homesrv (http://localhost:9100/mcp): handshake, three discovered tools with update_task correctly NOT read-only, a live list_projects call, a tool excluded by allow_tools refused, and the same server refused outright once allow_private was dropped. Tests cover both transports (the stdio one against a real subprocess), SSE and JSON framing, session echo, reconnect, and the config validation. --- internal/config/config.go | 126 ++++++++ internal/config/mcp_test.go | 82 ++++++ internal/mcp/allowlist.go | 61 ++++ internal/mcp/client.go | 267 +++++++++++++++++ internal/mcp/http.go | 154 ++++++++++ internal/mcp/jsonrpc.go | 72 +++++ internal/mcp/manager.go | 431 ++++++++++++++++++++++++++++ internal/mcp/mcp_test.go | 446 +++++++++++++++++++++++++++++ internal/mcp/stdio.go | 149 ++++++++++ internal/mcp/stdio_test.go | 149 ++++++++++ internal/mcp/webfetchdoor.go | 46 +++ internal/webfetch/webfetch.go | 59 +++- internal/webfetch/webfetch_test.go | 72 +++++ 13 files changed, 2108 insertions(+), 6 deletions(-) create mode 100644 internal/config/mcp_test.go create mode 100644 internal/mcp/allowlist.go create mode 100644 internal/mcp/client.go create mode 100644 internal/mcp/http.go create mode 100644 internal/mcp/jsonrpc.go create mode 100644 internal/mcp/manager.go create mode 100644 internal/mcp/mcp_test.go create mode 100644 internal/mcp/stdio.go create mode 100644 internal/mcp/stdio_test.go create mode 100644 internal/mcp/webfetchdoor.go diff --git a/internal/config/config.go b/internal/config/config.go index 5d9a0dd..f3a0340 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -22,6 +22,7 @@ import ( "github.com/kami/maven/internal/delivery/ntfysink" "github.com/kami/maven/internal/delivery/telegramsink" + "github.com/kami/maven/internal/mcp" "github.com/kami/maven/internal/morning" "github.com/kami/maven/internal/update" "github.com/robfig/cron/v3" @@ -191,6 +192,119 @@ type Config struct { // discovers and executes capabilities through Hexis for ecosystem actions. // nil ⇒ no capability-aware routing. Hexis *HexisConfig `json:"hexis,omitempty"` + + // MCP — Model Context Protocol servers Maven connects OUT to (Vikunja + // #251). nil / absent / no enabled server ⇒ no connection is made and no + // tool is discovered, like every other capability that reaches outside the + // box. She is a client here, never a server: nothing exposes her own + // capabilities to an outside caller. See MCPConfig. + MCP *MCPConfig `json:"mcp,omitempty"` +} + +// MCPConfig — the MCP client block. Servers are dark until one has +// `"enabled": true`, and a discovered tool is only ever PROPOSED: Kami enables +// it on /tools, on the authed surface, exactly as he would a shell tool. The +// voice path can never grant a capability to itself. +type MCPConfig struct { + // Servers — the configured servers. Each needs exactly one of command + // (a subprocess on this box) or url (a streamable-HTTP endpoint). + Servers []MCPServerConfig `json:"servers,omitempty"` + + // Timeout — per-call budget for every server that does not set its own. + // 0 ⇒ mcp.DefaultTimeout (15s). A tool slower than this is not usable in a + // spoken turn. + Timeout Duration `json:"timeout,omitempty"` + + // AllowHosts / DenyHosts — the host lists for the shared webfetch door that + // url servers go through. Deny wins. Private addresses are refused + // unconditionally unless the individual server sets allow_private. + AllowHosts []string `json:"allow_hosts,omitempty"` + DenyHosts []string `json:"deny_hosts,omitempty"` + + // MaxBytes — cap on one JSON-RPC response. 0 ⇒ webfetch.DefaultMaxBytes. + MaxBytes int64 `json:"max_bytes,omitempty"` +} + +// MCPServerConfig — one MCP server. +type MCPServerConfig struct { + // Name — the local handle. It prefixes every tool this server contributes + // ("vikunja" + "list_tasks" ⇒ the allowlist row "vikunja_list_tasks") and + // becomes the store scope "mcp:", so its provenance is readable on + // /tools without opening the config. + Name string `json:"name"` + + // Command / Args / Env / Dir — a stdio server: a child process of mavend, + // on this box, under this user. argv, never a shell string. + Command string `json:"command,omitempty"` + Args []string `json:"args,omitempty"` + Env []string `json:"env,omitempty"` + Dir string `json:"dir,omitempty"` + + // URL — a streamable-HTTP endpoint. It is fetched through + // internal/webfetch, so the SSRF guard, the redirect cap, the size cap and + // the one-request-per-host-per-second limit all apply. + URL string `json:"url,omitempty"` + + // AllowPrivate — let THIS server be a loopback or LAN address. The Vikunja + // server on homesrv is "http://localhost:9100/mcp", which is refused + // without this flag. Understand what it means before setting it: a local + // server is a DIFFERENT trust level from a public one. It is inside the + // network, it usually needs no credential, and it can change things that + // matter — so an argument the router got wrong lands somewhere real. Set it + // only for a server you run yourself, and prefer allow_tools with it. + AllowPrivate bool `json:"allow_private,omitempty"` + + // AllowTools — when set, the ONLY remote tool names taken from this server. + // This is the knob that keeps the catalogue deliberate: the resident model + // is a 1.7B with a 4096-token context, and a tool name it half-remembers is + // a wrong act, so fewer and better-chosen beats complete. + AllowTools []string `json:"allow_tools,omitempty"` + + // MaxTools — cap on this server's contribution. 0 ⇒ mcp.DefaultMaxTools (12). + MaxTools int `json:"max_tools,omitempty"` + + // Timeout — per-call budget for this server. 0 ⇒ MCPConfig.Timeout. + Timeout Duration `json:"timeout,omitempty"` + + // Enabled — false (the default) keeps a configured server described but + // dark, so a block can be written and reviewed before it is switched on. + Enabled bool `json:"enabled,omitempty"` +} + +// MCPServers maps the config blocks onto the mcp package's own type. It lives +// here so config validation and daemon wiring cannot drift on the mapping. +// Returns nil when nothing is configured or nothing is enabled. +func (c *Config) MCPServers() []mcp.ServerConfig { + if c.MCP == nil { + return nil + } + out := make([]mcp.ServerConfig, 0, len(c.MCP.Servers)) + for _, s := range c.MCP.Servers { + if !s.Enabled { + continue + } + timeout := time.Duration(s.Timeout) + if timeout <= 0 { + timeout = time.Duration(c.MCP.Timeout) + } + out = append(out, mcp.ServerConfig{ + Name: s.Name, + Command: s.Command, + Args: s.Args, + Env: s.Env, + Dir: s.Dir, + URL: s.URL, + AllowPrivate: s.AllowPrivate, + AllowTools: s.AllowTools, + MaxTools: s.MaxTools, + Timeout: timeout, + Enabled: true, + }) + } + if len(out) == 0 { + return nil + } + return out } // PraxisConfig — maven's connection to the Praxis attention service. @@ -783,6 +897,12 @@ func (c *Config) applyDefaults() { c.Feeds = nil } + // Same rule for MCP: a block with no server, or none enabled, is the same + // as no block at all. Normalising it to nil keeps "off" in one place. + if c.MCP != nil && len(c.MCPServers()) == 0 { + c.MCP = nil + } + // Same rule for the crawler: a block that neither answers on demand nor // watches anything has nothing to do, so it is normalised to "off". if c.Crawl != nil && !c.Crawl.OnDemand && len(c.Crawl.Watches) == 0 { @@ -887,6 +1007,12 @@ func (c *Config) validate() error { return fmt.Errorf("routine %q: bad cron %q: %w", r.Name, r.Cron, err) } } + // An MCP block with a typo (no name, both command and url, a bare hostname + // as the url) fails here, at startup, rather than at the first turn that + // needed the tool. + if err := mcp.Validate(c.MCPServers()); err != nil { + return err + } if len(c.MorningRoutines) > 0 { if err := morning.Validate(morningRoutinesFromConfig(c.MorningRoutines)); err != nil { return err diff --git a/internal/config/mcp_test.go b/internal/config/mcp_test.go new file mode 100644 index 0000000..48e05e2 --- /dev/null +++ b/internal/config/mcp_test.go @@ -0,0 +1,82 @@ +package config + +import ( + "testing" + "time" +) + +func TestMCPAbsentIsOff(t *testing.T) { + c, err := Load(writeConfig(t, `{}`)) + if err != nil { + t.Fatal(err) + } + if c.MCP != nil { + t.Error("no mcp block ⇒ nil") + } + if got := c.MCPServers(); got != nil { + t.Errorf("MCPServers() = %+v, want nil", got) + } +} + +// A described-but-not-enabled server must not be wired. This is how a block can +// sit in the config file, reviewed, before it is switched on. +func TestMCPDisabledServerIsOff(t *testing.T) { + c, err := Load(writeConfig(t, `{"mcp":{"servers":[ + {"name":"vikunja","url":"http://localhost:9100/mcp","allow_private":true}]}}`)) + if err != nil { + t.Fatal(err) + } + if c.MCP != nil { + t.Errorf("a block with nothing enabled must normalise to nil, got %+v", c.MCP) + } + if got := c.MCPServers(); len(got) != 0 { + t.Errorf("MCPServers() = %+v", got) + } +} + +func TestMCPEnabledServerMapping(t *testing.T) { + c, err := Load(writeConfig(t, `{"mcp":{ + "timeout":"5s", + "servers":[ + {"name":"vikunja","url":"http://localhost:9100/mcp","allow_private":true, + "allow_tools":["list_tasks"],"max_tools":3,"enabled":true}, + {"name":"files","command":"mcp-server-fs","args":["/srv"],"timeout":"1s","enabled":true}, + {"name":"off","command":"nope"} + ]}}`)) + if err != nil { + t.Fatal(err) + } + got := c.MCPServers() + if len(got) != 2 { + t.Fatalf("servers = %+v", got) + } + if got[0].Name != "vikunja" || !got[0].AllowPrivate || got[0].MaxTools != 3 || + len(got[0].AllowTools) != 1 || got[0].Timeout != 5*time.Second { + t.Errorf("vikunja mapped wrong: %+v", got[0]) + } + if got[1].Command != "mcp-server-fs" || len(got[1].Args) != 1 || got[1].Timeout != time.Second { + t.Errorf("files mapped wrong: %+v", got[1]) + } + // allow_private is per server and must not leak to the other one. + if got[1].AllowPrivate { + t.Error("allow_private leaked between servers") + } +} + +func TestMCPBadServerFailsAtStartup(t *testing.T) { + cases := map[string]string{ + "no name": `{"mcp":{"servers":[{"command":"x","enabled":true}]}}`, + "both": `{"mcp":{"servers":[{"name":"a","command":"x","url":"http://a.test","enabled":true}]}}`, + "neither": `{"mcp":{"servers":[{"name":"a","enabled":true}]}}`, + "bad scheme": `{"mcp":{"servers":[{"name":"a","url":"unix:///run/x.sock","enabled":true}]}}`, + "duplicate": `{"mcp":{"servers":[{"name":"a","command":"x","enabled":true},{"name":"a","command":"y","enabled":true}]}}`, + "spacey name": `{"mcp":{"servers":[{"name":"a b","command":"x","enabled":true}]}}`, + } + for name, body := range cases { + t.Run(name, func(t *testing.T) { + if _, err := Load(writeConfig(t, body)); err == nil { + t.Fatal("want a startup error") + } + }) + } +} diff --git a/internal/mcp/allowlist.go b/internal/mcp/allowlist.go new file mode 100644 index 0000000..09c9d06 --- /dev/null +++ b/internal/mcp/allowlist.go @@ -0,0 +1,61 @@ +package mcp + +import ( + "regexp" + "strings" +) + +// CmdPrefix is the reserved first argv element that marks an allowlist row as +// an MCP call rather than a process. An MCP tool row looks like +// +// name: "vikunja_list_tasks" cmd: ["mcp", "vikunja", "list_tasks"] +// +// which is why there is no new column and no migration: the store, the /tools +// page, ProposeTool, EnableTool, DisableTool, the act matcher and the confirm +// turn all keep working unchanged. The executor is the only place that has to +// know the difference, and it is one branch on Cmd[0]. +// +// The rest of the allowlist discipline is inherited whole: a row that is not +// status='enabled' does not run, and a row marked destructive does not run on +// first hearing. Nothing here can enable itself — discovery only proposes. +const CmdPrefix = "mcp" + +// Cmd builds the argv encoding for a discovered tool. +func Cmd(server, tool string) []string { return []string{CmdPrefix, server, tool} } + +// ParseCmd recognises an MCP allowlist row. ok=false for an ordinary process +// tool, which is what almost every row is. +func ParseCmd(cmd []string) (server, tool string, ok bool) { + if len(cmd) != 3 || cmd[0] != CmdPrefix { + return "", "", false + } + if cmd[1] == "" || cmd[2] == "" { + return "", "", false + } + return cmd[1], cmd[2], true +} + +var notName = regexp.MustCompile(`[^a-z0-9_]+`) + +// LocalName is the allowlist name for a discovered tool: the server handle, an +// underscore, the remote name, lowercased and stripped of anything that is not +// a word character. Namespacing by server is what keeps two servers that both +// offer "search" from colliding, and what makes the provenance of a row on the +// /tools page obvious without opening the diff. +func LocalName(server, tool string) string { + clean := func(s string) string { + return strings.Trim(notName.ReplaceAllString(strings.ToLower(strings.TrimSpace(s)), "_"), "_") + } + s, t := clean(server), clean(tool) + switch { + case s == "": + return t + case t == "": + return s + } + return s + "_" + t +} + +// Scope is the store scope for a server's rows, so the /tools page can group +// them and a human can tell at a glance where a capability came from. +func Scope(server string) string { return "mcp:" + server } diff --git a/internal/mcp/client.go b/internal/mcp/client.go new file mode 100644 index 0000000..826c3f1 --- /dev/null +++ b/internal/mcp/client.go @@ -0,0 +1,267 @@ +package mcp + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + "sync/atomic" +) + +// Errors callers distinguish. +var ( + // ErrClosed — the transport is gone (subprocess died, client closed). + ErrClosed = errors.New("mcp: connection is closed") + // ErrNotInitialized — a call was made before the initialize handshake. + ErrNotInitialized = errors.New("mcp: not initialized") + // ErrToolFailed — the server ran the tool and reported an error result. + ErrToolFailed = errors.New("mcp: tool reported an error") +) + +// Tool is one tool a server offers, in the form Maven cares about. +// +// ReadOnly comes from the server's own readOnlyHint annotation and decides +// whether the allowlist row is marked destructive: no hint, or a false one, +// means "assume it mutates", which routes the call through the confirm turn. +// Guessing wrong in that direction only costs a question. +type Tool struct { + Server string + Name string + Description string + InputSchema json.RawMessage + ReadOnly bool +} + +// Resource is one resource a server offers. Contents are fetched separately — +// listing is cheap, reading is not. +type Resource struct { + Server string + URI string + Name string + MIMEType string +} + +// ServerInfo is what came back from the handshake. +type ServerInfo struct { + Name string `json:"name"` + Version string `json:"version"` + ProtocolVersion string `json:"-"` +} + +// Client is one connected MCP server. Safe for concurrent use. +type Client struct { + name string + tr transport + next atomic.Int64 + + mu sync.Mutex + info ServerInfo + ready bool +} + +// newClient wraps a transport. Callers use Dial* in manager.go. +func newClient(name string, tr transport) *Client { + return &Client{name: name, tr: tr} +} + +// Name — the local name of this server (the config key, not the server's own). +func (c *Client) Name() string { return c.name } + +// Info — what the server said about itself during the handshake. +func (c *Client) Info() ServerInfo { + c.mu.Lock() + defer c.mu.Unlock() + return c.info +} + +// Initialize performs the MCP handshake and sends notifications/initialized. +// Capabilities we declare are empty on purpose: Maven consumes, she does not +// offer sampling or roots back to the server. +func (c *Client) Initialize(ctx context.Context) error { + var out struct { + ProtocolVersion string `json:"protocolVersion"` + ServerInfo ServerInfo `json:"serverInfo"` + } + err := c.call(ctx, "initialize", map[string]any{ + "protocolVersion": ProtocolVersion, + "capabilities": map[string]any{}, + "clientInfo": map[string]any{"name": "maven", "version": "1.0"}, + }, &out) + if err != nil { + return err + } + if strings.TrimSpace(out.ProtocolVersion) == "" { + return fmt.Errorf("mcp: %s: handshake returned no protocol version", c.name) + } + out.ServerInfo.ProtocolVersion = out.ProtocolVersion + c.mu.Lock() + c.info, c.ready = out.ServerInfo, true + c.mu.Unlock() + // Best effort: a stateless HTTP server may not care, and a failure here is + // not worth dropping a working connection over. + _ = c.tr.Notify(ctx, "notifications/initialized", map[string]any{}) + return nil +} + +// ListTools discovers the server's tools. +func (c *Client) ListTools(ctx context.Context) ([]Tool, error) { + if !c.initialized() { + return nil, ErrNotInitialized + } + var out struct { + Tools []struct { + Name string `json:"name"` + Description string `json:"description"` + InputSchema json.RawMessage `json:"inputSchema"` + Annotations *struct { + ReadOnlyHint bool `json:"readOnlyHint"` + } `json:"annotations"` + } `json:"tools"` + } + if err := c.call(ctx, "tools/list", map[string]any{}, &out); err != nil { + return nil, err + } + tools := make([]Tool, 0, len(out.Tools)) + for _, t := range out.Tools { + if strings.TrimSpace(t.Name) == "" { + continue + } + tools = append(tools, Tool{ + Server: c.name, + Name: t.Name, + Description: strings.TrimSpace(t.Description), + InputSchema: t.InputSchema, + ReadOnly: t.Annotations != nil && t.Annotations.ReadOnlyHint, + }) + } + return tools, nil +} + +// CallTool runs one tool and returns its text content, joined by newlines. +// Non-text content (images, blobs) is dropped: everything downstream of here +// is a spoken or written sentence. +// +// args is exactly what the router produced. Nothing else — no history, no +// notes, no persona — is in scope here, by construction. +func (c *Client) CallTool(ctx context.Context, name string, args map[string]any) (string, error) { + if !c.initialized() { + return "", ErrNotInitialized + } + if args == nil { + args = map[string]any{} + } + var out struct { + IsError bool `json:"isError"` + Content []struct { + Type string `json:"type"` + Text string `json:"text"` + } `json:"content"` + } + if err := c.call(ctx, "tools/call", map[string]any{"name": name, "arguments": args}, &out); err != nil { + return "", err + } + var parts []string + for _, ct := range out.Content { + if ct.Type == "text" && strings.TrimSpace(ct.Text) != "" { + parts = append(parts, strings.TrimSpace(ct.Text)) + } + } + text := strings.Join(parts, "\n") + if out.IsError { + return text, fmt.Errorf("%w: %s/%s: %s", ErrToolFailed, c.name, name, text) + } + return text, nil +} + +// ListResources discovers the server's resources. A server without the +// resources capability answers with an error; that is not fatal, the caller +// gets an empty list. +func (c *Client) ListResources(ctx context.Context) ([]Resource, error) { + if !c.initialized() { + return nil, ErrNotInitialized + } + var out struct { + Resources []struct { + URI string `json:"uri"` + Name string `json:"name"` + MIMEType string `json:"mimeType"` + } `json:"resources"` + } + if err := c.call(ctx, "resources/list", map[string]any{}, &out); err != nil { + return nil, err + } + res := make([]Resource, 0, len(out.Resources)) + for _, r := range out.Resources { + if strings.TrimSpace(r.URI) == "" { + continue + } + res = append(res, Resource{Server: c.name, URI: r.URI, Name: r.Name, MIMEType: r.MIMEType}) + } + return res, nil +} + +// ReadResource returns a resource's text contents, joined by newlines. This is +// the RAG-hint path: the text can be pasted into a router or phraser prompt. +func (c *Client) ReadResource(ctx context.Context, uri string) (string, error) { + if !c.initialized() { + return "", ErrNotInitialized + } + var out struct { + Contents []struct { + Text string `json:"text"` + } `json:"contents"` + } + if err := c.call(ctx, "resources/read", map[string]any{"uri": uri}, &out); err != nil { + return "", err + } + var parts []string + for _, ct := range out.Contents { + if strings.TrimSpace(ct.Text) != "" { + parts = append(parts, strings.TrimSpace(ct.Text)) + } + } + return strings.Join(parts, "\n"), nil +} + +// Close drops the connection. +func (c *Client) Close() error { + c.mu.Lock() + c.ready = false + c.mu.Unlock() + return c.tr.Close() +} + +func (c *Client) initialized() bool { + c.mu.Lock() + defer c.mu.Unlock() + return c.ready +} + +// alive reports whether the underlying transport can still carry a call. HTTP +// is stateless, so it is always alive; a dead subprocess is not. +func (c *Client) alive() bool { + if s, ok := c.tr.(*stdioTransport); ok { + return s.alive() + } + return true +} + +func (c *Client) call(ctx context.Context, method string, params any, out any) error { + req := &rpcRequest{JSONRPC: "2.0", ID: c.next.Add(1), Method: method, Params: params} + resp, err := c.tr.Call(ctx, req) + if err != nil { + return fmt.Errorf("mcp: %s: %s: %w", c.name, method, err) + } + if resp.Error != nil { + return fmt.Errorf("mcp: %s: %s: %w", c.name, method, resp.Error) + } + if out == nil || len(resp.Result) == 0 { + return nil + } + if err := json.Unmarshal(resp.Result, out); err != nil { + return fmt.Errorf("mcp: %s: %s: decode result: %w", c.name, method, err) + } + return nil +} diff --git a/internal/mcp/http.go b/internal/mcp/http.go new file mode 100644 index 0000000..a717266 --- /dev/null +++ b/internal/mcp/http.go @@ -0,0 +1,154 @@ +package mcp + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "sync" +) + +// Poster is the HTTP seam: internal/webfetch.Fetcher satisfies it. The +// transport takes it as an interface so a test can serve a fake without a +// listener, and so that the ONLY implementation wired in production is the +// guarded fetcher — an MCP endpoint cannot get a bare http.Client this way. +type Poster interface { + Post(ctx context.Context, rawURL, contentType string, body []byte, hdr map[string]string) (*PostResponse, error) +} + +// PostResponse is the shape webfetch returns, restated here so this package +// does not depend on it structurally. +type PostResponse struct { + Status int + ContentType string + Body []byte + Header map[string]string +} + +// httpTransport speaks streamable HTTP: every request is a POST to one +// endpoint, and the reply is either a JSON object or a text/event-stream frame +// carrying one. Both are accepted — servers pick per response, and the two the +// LAN runs disagree about which. +type httpTransport struct { + poster Poster + url string + + mu sync.Mutex + session string // Mcp-Session-Id, echoed back when the server issues one +} + +func newHTTPTransport(post Poster, endpoint string) *httpTransport { + return &httpTransport{poster: post, url: endpoint} +} + +func (t *httpTransport) Call(ctx context.Context, req *rpcRequest) (*rpcResponse, error) { + body, err := t.send(ctx, req) + if err != nil { + return nil, err + } + frame, err := decodeFrame(body) + if err != nil { + return nil, err + } + var resp rpcResponse + if err := json.Unmarshal(frame, &resp); err != nil { + return nil, fmt.Errorf("mcp: decode response: %w", err) + } + return &resp, nil +} + +func (t *httpTransport) Notify(ctx context.Context, method string, params any) error { + _, err := t.send(ctx, &rpcRequest{JSONRPC: "2.0", Method: method, Params: params}) + return err +} + +func (t *httpTransport) send(ctx context.Context, req *rpcRequest) ([]byte, error) { + req.JSONRPC = "2.0" + raw, err := json.Marshal(req) + if err != nil { + return nil, err + } + hdr := map[string]string{"Accept": "application/json, text/event-stream"} + t.mu.Lock() + if t.session != "" { + hdr["Mcp-Session-Id"] = t.session + } + t.mu.Unlock() + + resp, err := t.poster.Post(ctx, t.url, "application/json", raw, hdr) + if err != nil { + return nil, err + } + if sid := headerGet(resp.Header, "Mcp-Session-Id"); sid != "" { + t.mu.Lock() + t.session = sid + t.mu.Unlock() + } + return resp.Body, nil +} + +func (t *httpTransport) Close() error { + t.mu.Lock() + t.session = "" + t.mu.Unlock() + return nil +} + +func headerGet(h map[string]string, key string) string { + if h == nil { + return "" + } + if v, ok := h[key]; ok { + return v + } + lower := strings.ToLower(key) + for k, v := range h { + if strings.ToLower(k) == lower { + return v + } + } + return "" +} + +// decodeFrame pulls the JSON object out of a body that is either raw JSON or +// SSE. For SSE we take the LAST data: payload that parses, which is the +// response — earlier frames on the stream are progress notifications. +func decodeFrame(body []byte) ([]byte, error) { + trimmed := bytes.TrimSpace(body) + if len(trimmed) == 0 { + return nil, errors.New("mcp: empty response body") + } + if trimmed[0] == '{' || trimmed[0] == '[' { + return trimmed, nil + } + var last []byte + sc := bufio.NewScanner(bytes.NewReader(trimmed)) + sc.Buffer(make([]byte, 0, 64<<10), maxLine) + for sc.Scan() { + line := strings.TrimSpace(sc.Text()) + if !strings.HasPrefix(line, "data:") { + continue + } + payload := strings.TrimSpace(strings.TrimPrefix(line, "data:")) + if payload == "" { + continue + } + var probe map[string]json.RawMessage + if json.Unmarshal([]byte(payload), &probe) != nil { + continue + } + if _, isResp := probe["id"]; isResp { + last = []byte(payload) + } + } + if err := sc.Err(); err != nil { + return nil, fmt.Errorf("mcp: read event stream: %w", err) + } + if last == nil { + return nil, errors.New("mcp: no JSON-RPC response in event stream") + } + return last, nil +} diff --git a/internal/mcp/jsonrpc.go b/internal/mcp/jsonrpc.go new file mode 100644 index 0000000..45222cf --- /dev/null +++ b/internal/mcp/jsonrpc.go @@ -0,0 +1,72 @@ +// Package mcp is Maven's Model Context Protocol CLIENT. She is a host: she +// connects OUT to MCP servers, discovers the tools and resources they offer, +// and hands them to the parts of her that already exist for this — the tool +// allowlist in the store, the confirm turn for anything that mutates, the +// stage-3 gate that makes an uncertain act ask instead of run. +// +// She is not an MCP server. Nothing here exposes her own capabilities to an +// outside caller; docs/plans/06-mcp-support.md asks for the host direction only. +// +// Boundaries, in code rather than in prose: +// +// - OFF unless configured. No mcp_servers block ⇒ no manager, no goroutine, +// no socket. +// - A remote server is reached through internal/webfetch, so the SSRF guard, +// the size cap, the redirect cap and the per-host rate limit all apply to +// an MCP endpoint exactly as they do to a news feed. Reaching a loopback +// or LAN server means explicitly setting allow_private on THAT server — +// a different trust level, spelled out per server rather than globally. +// - Only the tool name and the arguments the router produced are sent. This +// package never sees his notes, facts, history or the persona block, and +// has no API through which a caller could pass them. +// - Discovery proposes, it does not enable. A discovered tool lands as a +// 'proposed' row; a human enables it on the authed surface. +package mcp + +import ( + "context" + "encoding/json" + "fmt" +) + +// ProtocolVersion — the spec revision we ask for in the initialize handshake. +// A server that answers with a different one is accepted (the spec says the +// client may proceed if it can support what came back); we only refuse when it +// answers with no version at all, which means it is not an MCP server. +const ProtocolVersion = "2025-06-18" + +// rpcRequest / rpcResponse — JSON-RPC 2.0. Deliberately hand-rolled: the wire +// format is four fields, and the repo vendors its dependencies, so pulling a +// library in for this would cost more than it saves. +type rpcRequest struct { + JSONRPC string `json:"jsonrpc"` + ID int64 `json:"id,omitempty"` + Method string `json:"method"` + Params any `json:"params,omitempty"` +} + +type rpcResponse struct { + JSONRPC string `json:"jsonrpc"` + ID *int64 `json:"id"` + Result json.RawMessage `json:"result,omitempty"` + Error *rpcError `json:"error,omitempty"` +} + +type rpcError struct { + Code int `json:"code"` + Message string `json:"message"` +} + +func (e *rpcError) Error() string { return fmt.Sprintf("mcp: rpc error %d: %s", e.Code, e.Message) } + +// transport carries one JSON-RPC conversation. Implementations: stdioTransport +// (a subprocess on this box) and httpTransport (streamable HTTP, guarded by +// webfetch). Both must be safe for concurrent use by the Client. +type transport interface { + // Call sends a request and returns the matching response. + Call(ctx context.Context, req *rpcRequest) (*rpcResponse, error) + // Notify sends a notification (no id, no reply expected). + Notify(ctx context.Context, method string, params any) error + // Close releases the transport (kills the subprocess, drops the session). + Close() error +} diff --git a/internal/mcp/manager.go b/internal/mcp/manager.go new file mode 100644 index 0000000..edeb017 --- /dev/null +++ b/internal/mcp/manager.go @@ -0,0 +1,431 @@ +package mcp + +import ( + "context" + "errors" + "fmt" + "log" + "sort" + "strings" + "sync" + "time" +) + +// Defaults for a server block. Small numbers on purpose — see MaxTools. +const ( + // DefaultTimeout bounds one JSON-RPC call. A tool that takes longer than + // this is not usable in a spoken turn anyway. + DefaultTimeout = 15 * time.Second + // DefaultMaxTools caps how many tools ONE server may contribute. The + // resident model is a 1.7B with a 4096-token context: a catalogue of forty + // tool names does not fit in its head, and a name it half-remembers is a + // wrong act. Twelve per server is already generous. + DefaultMaxTools = 12 + // DefaultReconnectEvery is how long the manager waits before re-dialing a + // server whose connection died. + DefaultReconnectEvery = 30 * time.Second +) + +// ErrNoServer — the named server is not configured or not connected. +var ErrNoServer = errors.New("mcp: no such server") + +// ServerConfig is one configured MCP server. Off unless present. +// +// Exactly one of Command (a subprocess on this box) or URL (a remote or +// loopback HTTP endpoint) must be set. +type ServerConfig struct { + // Name is the local handle. It prefixes every tool this server + // contributes, so it must be short and a valid identifier-ish word. + Name string `json:"name"` + // Command + Args + Env + Dir describe a stdio server: a child process of + // mavend, on this box, under this user. argv, never a shell string. + Command string `json:"command,omitempty"` + Args []string `json:"args,omitempty"` + Env []string `json:"env,omitempty"` + Dir string `json:"dir,omitempty"` + // URL is a streamable-HTTP endpoint. It goes through internal/webfetch, so + // it inherits the SSRF guard, the size cap and the per-host rate limit. + URL string `json:"url,omitempty"` + // AllowPrivate lets THIS server be a loopback or LAN address + // (http://localhost:9100/mcp is the Vikunja server on homesrv). It is a + // per-server hole in the private-address guard and it is not the same trust + // level as a public endpoint: whatever is behind it is inside the network, + // so an argument the router got wrong reaches something that matters. Set + // it only for a server you run. + AllowPrivate bool `json:"allow_private,omitempty"` + // AllowTools, when non-empty, is the ONLY set of remote tool names taken + // from this server. This is the knob for keeping the catalogue small and + // deliberate rather than "whatever the server grew this week". + AllowTools []string `json:"allow_tools,omitempty"` + // MaxTools caps the contribution (0 ⇒ DefaultMaxTools). + MaxTools int `json:"max_tools,omitempty"` + // Timeout bounds one call (0 ⇒ DefaultTimeout). + Timeout time.Duration `json:"-"` + // Enabled=false keeps a configured server described but dark. + Enabled bool `json:"enabled"` +} + +// PosterFactory builds the HTTP door for one server. It is a factory rather +// than a single shared Poster because allow_private is per server: the fetcher +// that may reach http://localhost:9100/mcp must NOT be the same fetcher another +// server's public URL goes through, or one loopback exemption would quietly +// unlock the LAN for all of them. +type PosterFactory func(cfg ServerConfig) (Poster, error) + +// Manager owns the connections. Nothing here starts unless at least one server +// is configured and enabled. +type Manager struct { + newPoster PosterFactory + mu sync.Mutex + conns map[string]*conn + order []string +} + +type conn struct { + cfg ServerConfig + client *Client + tools []Tool + lastErr error + lastTry time.Time + dialedAt time.Time +} + +// NewManager builds a manager for the enabled servers in cfgs. newPoster is +// the guarded HTTP door factory for url servers; pass nil only when no url +// server is configured (a nil factory with a url server is reported per server +// at dial time rather than fatally, so one bad block never stops the daemon). +// +// Dialing is lazy: NewManager validates and records, Connect dials. +func NewManager(newPoster PosterFactory, cfgs []ServerConfig) (*Manager, error) { + m := &Manager{newPoster: newPoster, conns: map[string]*conn{}} + for _, c := range cfgs { + if !c.Enabled { + continue + } + if err := validate(c); err != nil { + return nil, err + } + if _, dup := m.conns[c.Name]; dup { + return nil, fmt.Errorf("mcp: duplicate server name %q", c.Name) + } + if c.Timeout <= 0 { + c.Timeout = DefaultTimeout + } + if c.MaxTools <= 0 { + c.MaxTools = DefaultMaxTools + } + m.conns[c.Name] = &conn{cfg: c} + m.order = append(m.order, c.Name) + } + sort.Strings(m.order) + return m, nil +} + +// Validate checks a set of server blocks without dialling anything, so a typo +// fails at startup rather than at the first turn that needed the tool. +func Validate(cfgs []ServerConfig) error { + seen := map[string]bool{} + for _, c := range cfgs { + if err := validate(c); err != nil { + return err + } + if seen[c.Name] { + return fmt.Errorf("mcp: duplicate server name %q", c.Name) + } + seen[c.Name] = true + } + return nil +} + +func validate(c ServerConfig) error { + if strings.TrimSpace(c.Name) == "" { + return errors.New("mcp: server needs a name") + } + if strings.ContainsAny(c.Name, " \t/:") { + return fmt.Errorf("mcp: server name %q must be one word without spaces, slashes or colons", c.Name) + } + hasCmd, hasURL := c.Command != "", c.URL != "" + if hasCmd == hasURL { + return fmt.Errorf("mcp: server %q needs exactly one of command or url", c.Name) + } + if hasURL && !strings.HasPrefix(c.URL, "http://") && !strings.HasPrefix(c.URL, "https://") { + return fmt.Errorf("mcp: server %q url must be http or https", c.Name) + } + return nil +} + +// Servers — the configured, enabled server names, sorted. +func (m *Manager) Servers() []string { + m.mu.Lock() + defer m.mu.Unlock() + return append([]string(nil), m.order...) +} + +// Empty reports whether nothing is configured. The daemon uses it to skip +// wiring entirely. +func (m *Manager) Empty() bool { + m.mu.Lock() + defer m.mu.Unlock() + return len(m.conns) == 0 +} + +// Connect dials every configured server, handshakes, and discovers tools. +// A server that fails is recorded and retried later by Refresh — one bad +// server never blocks the others, and never blocks boot. +func (m *Manager) Connect(ctx context.Context) { + for _, name := range m.Servers() { + if err := m.dial(ctx, name); err != nil { + log.Printf("mcp: %s: %v", name, err) + } + } +} + +func (m *Manager) dial(ctx context.Context, name string) error { + m.mu.Lock() + c, ok := m.conns[name] + if !ok { + m.mu.Unlock() + return ErrNoServer + } + cfg := c.cfg + c.lastTry = time.Now() + m.mu.Unlock() + + var tr transport + var err error + if cfg.Command != "" { + tr, err = newStdioTransport(ctx, append([]string{cfg.Command}, cfg.Args...), cfg.Env, cfg.Dir) + } else if m.newPoster == nil { + err = fmt.Errorf("server %q has a url but no http door was wired", name) + } else { + var poster Poster + if poster, err = m.newPoster(cfg); err == nil { + tr = newHTTPTransport(poster, cfg.URL) + } + } + if err != nil { + m.fail(name, err) + return err + } + + cl := newClient(name, tr) + ictx, cancel := context.WithTimeout(ctx, cfg.Timeout) + defer cancel() + if err := cl.Initialize(ictx); err != nil { + _ = cl.Close() + m.fail(name, err) + return err + } + tools, err := cl.ListTools(ictx) + if err != nil { + // A server with no tools capability is still a usable resource server. + log.Printf("mcp: %s: list tools: %v", name, err) + tools = nil + } + tools = filterTools(cfg, tools) + + m.mu.Lock() + if old := m.conns[name].client; old != nil { + _ = old.Close() + } + m.conns[name].client = cl + m.conns[name].tools = tools + m.conns[name].lastErr = nil + m.conns[name].dialedAt = time.Now() + m.mu.Unlock() + log.Printf("mcp: %s connected (%s %s), %d tool(s)", name, cl.Info().Name, cl.Info().Version, len(tools)) + return nil +} + +func (m *Manager) fail(name string, err error) { + m.mu.Lock() + defer m.mu.Unlock() + if c := m.conns[name]; c != nil { + c.lastErr = err + c.client = nil + c.tools = nil + } +} + +// filterTools applies AllowTools and MaxTools, and drops nameless entries. +// Sorted first, so the cap is deterministic rather than "whatever order the +// server felt like". +func filterTools(cfg ServerConfig, in []Tool) []Tool { + sort.Slice(in, func(i, j int) bool { return in[i].Name < in[j].Name }) + out := make([]Tool, 0, len(in)) + for _, t := range in { + if len(cfg.AllowTools) > 0 && !contains(cfg.AllowTools, t.Name) { + continue + } + out = append(out, t) + } + if cfg.MaxTools > 0 && len(out) > cfg.MaxTools { + log.Printf("mcp: %s offers %d tools, taking the first %d (raise max_tools or set allow_tools)", + cfg.Name, len(out), cfg.MaxTools) + out = out[:cfg.MaxTools] + } + return out +} + +func contains(hay []string, needle string) bool { + for _, h := range hay { + if h == needle { + return true + } + } + return false +} + +// Refresh re-dials any server that is down, if enough time has passed since the +// last attempt. Call it from the daemon's periodic tick — it is cheap when +// everything is up. +func (m *Manager) Refresh(ctx context.Context) { + now := time.Now() + var stale []string + m.mu.Lock() + for _, name := range m.order { + c := m.conns[name] + down := c.client == nil || !c.client.alive() + if down && now.Sub(c.lastTry) >= DefaultReconnectEvery { + stale = append(stale, name) + } + } + m.mu.Unlock() + for _, name := range stale { + if err := m.dial(ctx, name); err != nil { + log.Printf("mcp: %s: reconnect: %v", name, err) + } + } +} + +// Tools — every discovered tool across connected servers, sorted by +// server then name. +func (m *Manager) Tools() []Tool { + m.mu.Lock() + defer m.mu.Unlock() + var out []Tool + for _, name := range m.order { + out = append(out, m.conns[name].tools...) + } + return out +} + +// Status is one server's health, for the web surface. +type Status struct { + Name string + Transport string // "stdio" or "http" + Target string // command or url + Connected bool + Server string // the server's own name+version + Tools int + Err string +} + +// Status reports every configured server. +func (m *Manager) Status() []Status { + m.mu.Lock() + defer m.mu.Unlock() + out := make([]Status, 0, len(m.order)) + for _, name := range m.order { + c := m.conns[name] + s := Status{Name: name, Tools: len(c.tools)} + if c.cfg.Command != "" { + s.Transport, s.Target = "stdio", strings.Join(append([]string{c.cfg.Command}, c.cfg.Args...), " ") + } else { + s.Transport, s.Target = "http", c.cfg.URL + } + if c.client != nil { + s.Connected = true + s.Server = strings.TrimSpace(c.client.Info().Name + " " + c.client.Info().Version) + } + if c.lastErr != nil { + s.Err = c.lastErr.Error() + } + out = append(out, s) + } + return out +} + +// Call runs server's tool with args. Args come from the router and nothing +// else; there is no path here through which a note or a fact could travel. +func (m *Manager) Call(ctx context.Context, server, tool string, args map[string]any) (string, error) { + m.mu.Lock() + c := m.conns[server] + m.mu.Unlock() + if c == nil { + return "", fmt.Errorf("%w: %s", ErrNoServer, server) + } + m.mu.Lock() + cl, timeout, known := c.client, c.cfg.Timeout, false + for _, t := range c.tools { + if t.Name == tool { + known = true + break + } + } + m.mu.Unlock() + if cl == nil { + return "", fmt.Errorf("mcp: %s is not connected", server) + } + // The discovered-and-filtered set is the second allowlist: even an enabled + // store row cannot reach a tool the server stopped offering, or one + // allow_tools excludes. + if !known { + return "", fmt.Errorf("mcp: %s offers no tool %q", server, tool) + } + cctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + return cl.CallTool(cctx, tool, args) +} + +// Resources lists resources across connected servers. +func (m *Manager) Resources(ctx context.Context) []Resource { + m.mu.Lock() + clients := make([]*Client, 0, len(m.order)) + for _, name := range m.order { + if cl := m.conns[name].client; cl != nil { + clients = append(clients, cl) + } + } + m.mu.Unlock() + var out []Resource + for _, cl := range clients { + rs, err := cl.ListResources(ctx) + if err != nil { + continue // no resources capability; not an error worth logging per tick + } + out = append(out, rs...) + } + return out +} + +// ReadResource reads one resource from one server. +func (m *Manager) ReadResource(ctx context.Context, server, uri string) (string, error) { + m.mu.Lock() + c := m.conns[server] + var cl *Client + var timeout time.Duration + if c != nil { + cl, timeout = c.client, c.cfg.Timeout + } + m.mu.Unlock() + if cl == nil { + return "", fmt.Errorf("%w: %s", ErrNoServer, server) + } + cctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + return cl.ReadResource(cctx, uri) +} + +// Close shuts every connection down. +func (m *Manager) Close() error { + m.mu.Lock() + defer m.mu.Unlock() + for _, name := range m.order { + if cl := m.conns[name].client; cl != nil { + _ = cl.Close() + m.conns[name].client = nil + } + } + return nil +} diff --git a/internal/mcp/mcp_test.go b/internal/mcp/mcp_test.go new file mode 100644 index 0000000..3f29e6c --- /dev/null +++ b/internal/mcp/mcp_test.go @@ -0,0 +1,446 @@ +package mcp + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "sync" + "testing" + "time" +) + +// fakePoster answers POSTs from a canned handler, in either JSON or SSE form. +type fakePoster struct { + mu sync.Mutex + handler func(method string, params json.RawMessage) (any, *rpcError) + sse bool + session string + seen []map[string]string // headers of each request, for the session test + calls []string +} + +func (f *fakePoster) Post(_ context.Context, _, _ string, body []byte, hdr map[string]string) (*PostResponse, error) { + var req struct { + ID *int64 `json:"id"` + Method string `json:"method"` + Params json.RawMessage `json:"params"` + } + if err := json.Unmarshal(body, &req); err != nil { + return nil, err + } + f.mu.Lock() + f.seen = append(f.seen, hdr) + f.calls = append(f.calls, req.Method) + f.mu.Unlock() + + if req.ID == nil { // notification + return &PostResponse{Status: 202, Body: []byte(`{}`)}, nil + } + result, rerr := f.handler(req.Method, req.Params) + resp := map[string]any{"jsonrpc": "2.0", "id": *req.ID} + if rerr != nil { + resp["error"] = map[string]any{"code": rerr.Code, "message": rerr.Message} + } else { + resp["result"] = result + } + raw, _ := json.Marshal(resp) + out := &PostResponse{Status: 200, Body: raw, ContentType: "application/json", Header: map[string]string{}} + if f.sse { + out.ContentType = "text/event-stream" + out.Body = []byte("event: message\ndata: {\"jsonrpc\":\"2.0\",\"method\":\"notifications/progress\"}\n\nevent: message\ndata: " + string(raw) + "\n\n") + } + if f.session != "" { + out.Header["Mcp-Session-Id"] = f.session + } + return out, nil +} + +// echoServer is a handler with two tools, one read-only and one not. +func echoServer() func(string, json.RawMessage) (any, *rpcError) { + return func(method string, params json.RawMessage) (any, *rpcError) { + switch method { + case "initialize": + return map[string]any{ + "protocolVersion": ProtocolVersion, + "serverInfo": map[string]any{"name": "fake", "version": "0.1"}, + }, nil + case "tools/list": + return map[string]any{"tools": []any{ + map[string]any{ + "name": "read_thing", "description": "reads", + "inputSchema": map[string]any{"type": "object"}, + "annotations": map[string]any{"readOnlyHint": true}, + }, + map[string]any{"name": "break_thing", "description": "mutates"}, + }}, nil + case "tools/call": + var p struct { + Name string `json:"name"` + Args map[string]any `json:"arguments"` + } + _ = json.Unmarshal(params, &p) + if p.Name == "break_thing" { + return map[string]any{"isError": true, "content": []any{ + map[string]any{"type": "text", "text": "не вышло"}}}, nil + } + return map[string]any{"content": []any{ + map[string]any{"type": "text", "text": fmt.Sprintf("%s:%v", p.Name, p.Args["q"])}, + map[string]any{"type": "image", "text": "ignored"}, + }}, nil + case "resources/list": + return map[string]any{"resources": []any{ + map[string]any{"uri": "note://one", "name": "one", "mimeType": "text/plain"}, + map[string]any{"uri": "", "name": "nameless"}, + }}, nil + case "resources/read": + return map[string]any{"contents": []any{map[string]any{"text": "тело ресурса"}}}, nil + } + return nil, &rpcError{Code: -32601, Message: "method not found"} + } +} + +func dialFake(t *testing.T, p *fakePoster) *Client { + t.Helper() + c := newClient("fake", newHTTPTransport(p, "http://example.test/mcp")) + if err := c.Initialize(context.Background()); err != nil { + t.Fatalf("initialize: %v", err) + } + return c +} + +func TestHandshakeAndDiscovery(t *testing.T) { + for _, sse := range []bool{false, true} { + name := "json" + if sse { + name = "sse" + } + t.Run(name, func(t *testing.T) { + p := &fakePoster{handler: echoServer(), sse: sse} + c := dialFake(t, p) + if got := c.Info().Name; got != "fake" { + t.Fatalf("server name = %q", got) + } + if got := c.Info().ProtocolVersion; got != ProtocolVersion { + t.Fatalf("protocol = %q", got) + } + tools, err := c.ListTools(context.Background()) + if err != nil { + t.Fatalf("list tools: %v", err) + } + if len(tools) != 2 { + t.Fatalf("tools = %+v", tools) + } + byName := map[string]Tool{} + for _, tl := range tools { + byName[tl.Name] = tl + } + if !byName["read_thing"].ReadOnly { + t.Error("read_thing should be read-only (readOnlyHint true)") + } + // The important direction: no annotation ⇒ assume it mutates. + if byName["break_thing"].ReadOnly { + t.Error("break_thing has no readOnlyHint, must NOT be treated as read-only") + } + if byName["read_thing"].Server != "fake" { + t.Error("tool should carry its server handle") + } + }) + } +} + +func TestCallToolTextOnly(t *testing.T) { + c := dialFake(t, &fakePoster{handler: echoServer()}) + out, err := c.CallTool(context.Background(), "read_thing", map[string]any{"q": "привет"}) + if err != nil { + t.Fatalf("call: %v", err) + } + if out != "read_thing:привет" { + t.Fatalf("out = %q (non-text content must be dropped)", out) + } +} + +func TestCallToolErrorResult(t *testing.T) { + c := dialFake(t, &fakePoster{handler: echoServer()}) + out, err := c.CallTool(context.Background(), "break_thing", nil) + if err == nil { + t.Fatal("isError result must surface as an error") + } + if out != "не вышло" { + t.Fatalf("text should still come back, got %q", out) + } +} + +func TestResources(t *testing.T) { + c := dialFake(t, &fakePoster{handler: echoServer()}) + rs, err := c.ListResources(context.Background()) + if err != nil { + t.Fatalf("list resources: %v", err) + } + if len(rs) != 1 || rs[0].URI != "note://one" { + t.Fatalf("resources = %+v (a uri-less entry must be dropped)", rs) + } + body, err := c.ReadResource(context.Background(), "note://one") + if err != nil { + t.Fatalf("read: %v", err) + } + if body != "тело ресурса" { + t.Fatalf("body = %q", body) + } +} + +func TestCallBeforeInitializeRefused(t *testing.T) { + c := newClient("fake", newHTTPTransport(&fakePoster{handler: echoServer()}, "http://example.test/mcp")) + if _, err := c.CallTool(context.Background(), "read_thing", nil); err != ErrNotInitialized { + t.Fatalf("err = %v, want ErrNotInitialized", err) + } +} + +func TestSessionIDEchoed(t *testing.T) { + p := &fakePoster{handler: echoServer(), session: "sess-1"} + c := dialFake(t, p) + if _, err := c.ListTools(context.Background()); err != nil { + t.Fatal(err) + } + p.mu.Lock() + defer p.mu.Unlock() + last := p.seen[len(p.seen)-1] + if last["Mcp-Session-Id"] != "sess-1" { + t.Fatalf("session header not echoed: %+v", last) + } + if !strings.Contains(last["Accept"], "text/event-stream") { + t.Fatalf("Accept must offer both forms: %q", last["Accept"]) + } +} + +func TestHandshakeWithoutProtocolVersionRefused(t *testing.T) { + p := &fakePoster{handler: func(m string, _ json.RawMessage) (any, *rpcError) { + return map[string]any{"serverInfo": map[string]any{"name": "not-mcp"}}, nil + }} + c := newClient("x", newHTTPTransport(p, "http://example.test/mcp")) + if err := c.Initialize(context.Background()); err == nil { + t.Fatal("a reply with no protocolVersion is not an MCP server") + } +} + +func TestRPCErrorSurfaces(t *testing.T) { + c := dialFake(t, &fakePoster{handler: echoServer()}) + if _, err := c.callRaw(context.Background(), "nope/nope"); err == nil { + t.Fatal("want an rpc error") + } else if !strings.Contains(err.Error(), "method not found") { + t.Fatalf("err = %v", err) + } +} + +// callRaw is a test-only shim so the rpc-error path can be exercised without a +// typed wrapper for a method the server does not implement. +func (c *Client) callRaw(ctx context.Context, method string) (any, error) { + var out any + err := c.call(ctx, method, map[string]any{}, &out) + return out, err +} + +func TestDecodeFrame(t *testing.T) { + cases := []struct { + name, in, want string + wantErr bool + }{ + {name: "plain json", in: `{"id":1,"result":{}}`, want: `{"id":1,"result":{}}`}, + {name: "sse single", in: "event: message\ndata: {\"id\":1,\"result\":1}\n\n", want: `{"id":1,"result":1}`}, + { + name: "sse picks the response not the notification", + in: "data: {\"method\":\"notifications/progress\"}\n\ndata: {\"id\":2,\"result\":2}\n\n", + want: `{"id":2,"result":2}`, + }, + {name: "empty", in: " ", wantErr: true}, + {name: "sse with no response", in: "data: {\"method\":\"x\"}\n\n", wantErr: true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, err := decodeFrame([]byte(tc.in)) + if tc.wantErr { + if err == nil { + t.Fatalf("want error, got %q", got) + } + return + } + if err != nil { + t.Fatal(err) + } + if string(got) != tc.want { + t.Fatalf("got %q want %q", got, tc.want) + } + }) + } +} + +func TestValidate(t *testing.T) { + cases := []struct { + name string + cfg ServerConfig + wantErr bool + }{ + {name: "stdio ok", cfg: ServerConfig{Name: "a", Command: "echo"}}, + {name: "http ok", cfg: ServerConfig{Name: "a", URL: "http://x.test/mcp"}}, + {name: "no name", cfg: ServerConfig{Command: "echo"}, wantErr: true}, + {name: "spacey name", cfg: ServerConfig{Name: "a b", Command: "echo"}, wantErr: true}, + {name: "neither", cfg: ServerConfig{Name: "a"}, wantErr: true}, + {name: "both", cfg: ServerConfig{Name: "a", Command: "echo", URL: "http://x.test"}, wantErr: true}, + {name: "bad scheme", cfg: ServerConfig{Name: "a", URL: "file:///etc/passwd"}, wantErr: true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + err := Validate([]ServerConfig{tc.cfg}) + if (err != nil) != tc.wantErr { + t.Fatalf("err = %v, wantErr = %v", err, tc.wantErr) + } + }) + } + if err := Validate([]ServerConfig{{Name: "a", Command: "x"}, {Name: "a", Command: "y"}}); err == nil { + t.Error("duplicate names must be refused") + } +} + +func TestManagerOffWhenNothingEnabled(t *testing.T) { + m, err := NewManager(nil, []ServerConfig{{Name: "a", Command: "echo"}}) // Enabled=false + if err != nil { + t.Fatal(err) + } + if !m.Empty() { + t.Fatal("a server that is not enabled must not be wired") + } + m.Connect(context.Background()) + if got := m.Tools(); len(got) != 0 { + t.Fatalf("tools = %+v", got) + } +} + +func TestManagerDiscoversAndCalls(t *testing.T) { + p := &fakePoster{handler: echoServer()} + m, err := NewManager(func(ServerConfig) (Poster, error) { return p, nil }, + []ServerConfig{{Name: "fake", URL: "http://example.test/mcp", Enabled: true}}) + if err != nil { + t.Fatal(err) + } + m.Connect(context.Background()) + defer m.Close() + + tools := m.Tools() + if len(tools) != 2 { + t.Fatalf("tools = %+v", tools) + } + out, err := m.Call(context.Background(), "fake", "read_thing", map[string]any{"q": "да"}) + if err != nil { + t.Fatalf("call: %v", err) + } + if out != "read_thing:да" { + t.Fatalf("out = %q", out) + } + // The discovered set is a second allowlist. + if _, err := m.Call(context.Background(), "fake", "not_offered", nil); err == nil { + t.Error("a tool the server does not offer must be refused") + } + if _, err := m.Call(context.Background(), "other", "read_thing", nil); err == nil { + t.Error("an unconfigured server must be refused") + } + st := m.Status() + if len(st) != 1 || !st[0].Connected || st[0].Transport != "http" || st[0].Tools != 2 { + t.Fatalf("status = %+v", st) + } +} + +func TestManagerAllowToolsAndMaxTools(t *testing.T) { + p := &fakePoster{handler: echoServer()} + mk := func(cfg ServerConfig) *Manager { + cfg.Name, cfg.URL, cfg.Enabled = "fake", "http://example.test/mcp", true + m, err := NewManager(func(ServerConfig) (Poster, error) { return p, nil }, []ServerConfig{cfg}) + if err != nil { + t.Fatal(err) + } + m.Connect(context.Background()) + return m + } + m := mk(ServerConfig{AllowTools: []string{"read_thing"}}) + defer m.Close() + if got := m.Tools(); len(got) != 1 || got[0].Name != "read_thing" { + t.Fatalf("allow_tools ignored: %+v", got) + } + if _, err := m.Call(context.Background(), "fake", "break_thing", nil); err == nil { + t.Error("a tool excluded by allow_tools must be unreachable") + } + m2 := mk(ServerConfig{MaxTools: 1}) + defer m2.Close() + if got := m2.Tools(); len(got) != 1 || got[0].Name != "break_thing" { + t.Fatalf("max_tools should keep the first name-sorted tool: %+v", got) + } +} + +func TestManagerURLServerWithoutHTTPDoor(t *testing.T) { + m, err := NewManager(nil, []ServerConfig{{Name: "fake", URL: "http://example.test/mcp", Enabled: true}}) + if err != nil { + t.Fatal(err) + } + m.Connect(context.Background()) + st := m.Status() + if len(st) != 1 || st[0].Connected || st[0].Err == "" { + t.Fatalf("a url server with no poster must be recorded as failed: %+v", st) + } +} + +func TestManagerReconnectAfterFailure(t *testing.T) { + var mu sync.Mutex + fail := true + m, err := NewManager(func(ServerConfig) (Poster, error) { + mu.Lock() + defer mu.Unlock() + if fail { + return nil, fmt.Errorf("down") + } + return &fakePoster{handler: echoServer()}, nil + }, []ServerConfig{{Name: "fake", URL: "http://example.test/mcp", Enabled: true}}) + if err != nil { + t.Fatal(err) + } + defer m.Close() + m.Connect(context.Background()) + if m.Status()[0].Connected { + t.Fatal("should be down") + } + mu.Lock() + fail = false + mu.Unlock() + // Refresh honours the backoff, so pretend the last attempt was long ago. + m.mu.Lock() + m.conns["fake"].lastTry = time.Now().Add(-2 * DefaultReconnectEvery) + m.mu.Unlock() + m.Refresh(context.Background()) + if !m.Status()[0].Connected { + t.Fatalf("should have reconnected: %+v", m.Status()) + } +} + +func TestLocalNameAndCmd(t *testing.T) { + cases := [][3]string{ + {"vikunja", "list_tasks", "vikunja_list_tasks"}, + {"Vikunja", "Get Task Details", "vikunja_get_task_details"}, + {"fs", "read-file", "fs_read_file"}, + {"", "search", "search"}, + } + for _, c := range cases { + if got := LocalName(c[0], c[1]); got != c[2] { + t.Errorf("LocalName(%q,%q) = %q want %q", c[0], c[1], got, c[2]) + } + } + server, tool, ok := ParseCmd(Cmd("vikunja", "list_tasks")) + if !ok || server != "vikunja" || tool != "list_tasks" { + t.Fatalf("ParseCmd round-trip: %q %q %v", server, tool, ok) + } + for _, bad := range [][]string{nil, {"systemctl", "restart", "nginx"}, {"mcp", "vikunja"}, {"mcp", "", "x"}} { + if _, _, ok := ParseCmd(bad); ok { + t.Errorf("ParseCmd(%v) must not claim an ordinary tool row", bad) + } + } + if Scope("vikunja") != "mcp:vikunja" { + t.Error("scope") + } +} diff --git a/internal/mcp/stdio.go b/internal/mcp/stdio.go new file mode 100644 index 0000000..d597875 --- /dev/null +++ b/internal/mcp/stdio.go @@ -0,0 +1,149 @@ +package mcp + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "os/exec" + "strings" + "sync" +) + +// maxLine bounds one JSON-RPC frame from a subprocess. A tool result bigger +// than this is a misbehaving server, not something to buffer. +const maxLine = 1 << 20 // 1 MiB + +// stdioTransport speaks newline-delimited JSON-RPC to a child process. This is +// the local transport: the server runs on this box, under this user, and gets +// no network guard because it never touches the network on our behalf. +// +// Args are argv, never a shell string — the same discipline internal/tool +// keeps, for the same reason. +type stdioTransport struct { + mu sync.Mutex + cmd *exec.Cmd + in io.WriteCloser + out *bufio.Reader + dead bool +} + +func newStdioTransport(ctx context.Context, argv []string, env []string, dir string) (*stdioTransport, error) { + if len(argv) == 0 { + return nil, errors.New("mcp: stdio server needs a command") + } + cmd := exec.Command(argv[0], argv[1:]...) + cmd.Dir = dir + if len(env) > 0 { + cmd.Env = append(os.Environ(), env...) + } + cmd.Stderr = os.Stderr + in, err := cmd.StdinPipe() + if err != nil { + return nil, fmt.Errorf("mcp: stdin pipe: %w", err) + } + out, err := cmd.StdoutPipe() + if err != nil { + return nil, fmt.Errorf("mcp: stdout pipe: %w", err) + } + if err := cmd.Start(); err != nil { + return nil, fmt.Errorf("mcp: start %q: %w", argv[0], err) + } + return &stdioTransport{cmd: cmd, in: in, out: bufio.NewReaderSize(out, 64<<10)}, nil +} + +func (t *stdioTransport) Call(ctx context.Context, req *rpcRequest) (*rpcResponse, error) { + t.mu.Lock() + defer t.mu.Unlock() + if t.dead { + return nil, ErrClosed + } + if err := t.write(req); err != nil { + t.dead = true + return nil, err + } + // Read until the frame with our id turns up; anything else on the pipe is + // a notification or a server-initiated request we do not answer. + for { + if err := ctx.Err(); err != nil { + return nil, err + } + line, err := t.readLine() + if err != nil { + t.dead = true + return nil, err + } + var resp rpcResponse + if err := json.Unmarshal(line, &resp); err != nil { + continue // not a response frame; ignore rather than break the turn + } + if resp.ID == nil || *resp.ID != req.ID { + continue + } + return &resp, nil + } +} + +func (t *stdioTransport) Notify(ctx context.Context, method string, params any) error { + t.mu.Lock() + defer t.mu.Unlock() + if t.dead { + return ErrClosed + } + return t.write(&rpcRequest{JSONRPC: "2.0", Method: method, Params: params}) +} + +func (t *stdioTransport) write(req *rpcRequest) error { + req.JSONRPC = "2.0" + raw, err := json.Marshal(req) + if err != nil { + return err + } + if _, err := t.in.Write(append(raw, '\n')); err != nil { + return fmt.Errorf("mcp: write %s: %w", req.Method, err) + } + return nil +} + +func (t *stdioTransport) readLine() ([]byte, error) { + for { + line, err := t.out.ReadString('\n') + if err != nil { + if len(strings.TrimSpace(line)) == 0 { + return nil, fmt.Errorf("mcp: read: %w", err) + } + return []byte(line), nil + } + if len(line) > maxLine { + return nil, fmt.Errorf("mcp: frame exceeds %d bytes", maxLine) + } + if s := strings.TrimSpace(line); s != "" { + return []byte(s), nil + } + } +} + +func (t *stdioTransport) Close() error { + t.mu.Lock() + defer t.mu.Unlock() + t.dead = true + if t.in != nil { + _ = t.in.Close() + } + if t.cmd.Process != nil { + _ = t.cmd.Process.Kill() + _ = t.cmd.Wait() + } + return nil +} + +// alive reports whether the transport can still carry a call. The manager uses +// it to decide on a reconnect instead of retrying into a dead pipe. +func (t *stdioTransport) alive() bool { + t.mu.Lock() + defer t.mu.Unlock() + return !t.dead +} diff --git a/internal/mcp/stdio_test.go b/internal/mcp/stdio_test.go new file mode 100644 index 0000000..8e43b59 --- /dev/null +++ b/internal/mcp/stdio_test.go @@ -0,0 +1,149 @@ +package mcp + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "os" + "os/exec" + "strings" + "testing" +) + +// The stdio transport is tested against a real subprocess — this test binary, +// re-executed with MAVEN_MCP_FAKE set, acting as a minimal MCP server. No +// python, no fixture file, no network. +func TestMain(m *testing.M) { + if os.Getenv("MAVEN_MCP_FAKE") != "" { + fakeStdioServer() + return + } + os.Exit(m.Run()) +} + +func fakeStdioServer() { + h := echoServer() + sc := bufio.NewScanner(os.Stdin) + out := bufio.NewWriter(os.Stdout) + defer out.Flush() + for sc.Scan() { + line := strings.TrimSpace(sc.Text()) + if line == "" { + continue + } + var req struct { + ID *int64 `json:"id"` + Method string `json:"method"` + Params json.RawMessage `json:"params"` + } + if json.Unmarshal([]byte(line), &req) != nil { + continue + } + if req.ID == nil { + // A notification gets no reply, but we emit an unrelated + // notification so the client's frame-skipping is exercised. + _, _ = out.WriteString("{\"jsonrpc\":\"2.0\",\"method\":\"notifications/message\"}\n") + _ = out.Flush() + continue + } + result, rerr := h(req.Method, req.Params) + resp := map[string]any{"jsonrpc": "2.0", "id": *req.ID} + if rerr != nil { + resp["error"] = map[string]any{"code": rerr.Code, "message": rerr.Message} + } else { + resp["result"] = result + } + raw, _ := json.Marshal(resp) + _, _ = out.Write(append(raw, '\n')) + _ = out.Flush() + if os.Getenv("MAVEN_MCP_FAKE") == "die" && req.Method == "tools/list" { + return // hang up, so the reconnect path has something to see + } + } +} + +func stdioManager(t *testing.T, mode string) *Manager { + t.Helper() + self, err := os.Executable() + if err != nil { + t.Skipf("no executable path: %v", err) + } + if _, err := exec.LookPath(self); err != nil && !strings.Contains(self, "/") { + t.Skip("test binary not executable") + } + m, err := NewManager(nil, []ServerConfig{{ + Name: "fake", + Command: self, + Env: []string{"MAVEN_MCP_FAKE=" + mode}, + Enabled: true, + }}) + if err != nil { + t.Fatal(err) + } + m.Connect(context.Background()) + return m +} + +func TestStdioTransportEndToEnd(t *testing.T) { + m := stdioManager(t, "1") + defer m.Close() + st := m.Status() + if len(st) != 1 || !st[0].Connected { + t.Fatalf("status = %+v", st) + } + if st[0].Transport != "stdio" { + t.Fatalf("transport = %q", st[0].Transport) + } + if got := len(m.Tools()); got != 2 { + t.Fatalf("tools = %d", got) + } + out, err := m.Call(context.Background(), "fake", "read_thing", map[string]any{"q": "стдио"}) + if err != nil { + t.Fatalf("call: %v", err) + } + if out != "read_thing:стдио" { + t.Fatalf("out = %q", out) + } + res := m.Resources(context.Background()) + if len(res) != 1 || res[0].URI != "note://one" { + t.Fatalf("resources = %+v", res) + } + body, err := m.ReadResource(context.Background(), "fake", "note://one") + if err != nil { + t.Fatal(err) + } + if body != "тело ресурса" { + t.Fatalf("body = %q", body) + } +} + +func TestStdioServerThatDiesIsNotUsable(t *testing.T) { + m := stdioManager(t, "die") + defer m.Close() + // The server hung up after tools/list; the next call must fail cleanly + // rather than hang or panic. + if _, err := m.Call(context.Background(), "fake", "read_thing", nil); err == nil { + t.Fatal("a call into a dead server must error") + } +} + +func TestStdioMissingCommand(t *testing.T) { + m, err := NewManager(nil, []ServerConfig{{ + Name: "nope", Command: "/nonexistent/mcp-server-that-is-not-there", Enabled: true, + }}) + if err != nil { + t.Fatal(err) + } + m.Connect(context.Background()) + st := m.Status() + if st[0].Connected || st[0].Err == "" { + t.Fatalf("a missing binary must be recorded, not fatal: %+v", st) + } + if got := len(m.Tools()); got != 0 { + t.Fatalf("tools = %d", got) + } + if !strings.Contains(fmt.Sprint(st[0].Err), "start") { + t.Logf("err = %q", st[0].Err) + } +} diff --git a/internal/mcp/webfetchdoor.go b/internal/mcp/webfetchdoor.go new file mode 100644 index 0000000..3a892a2 --- /dev/null +++ b/internal/mcp/webfetchdoor.go @@ -0,0 +1,46 @@ +package mcp + +import ( + "context" + "fmt" + + "github.com/kami/maven/internal/webfetch" +) + +// WebfetchDoor builds the PosterFactory used in production: one guarded +// webfetch.Fetcher per url server, with that server's allow_private and the +// shared host lists and limits. +// +// One fetcher PER server is the point. allow_private is a hole in the +// private-address guard, and a hole punched for the Vikunja server on loopback +// must not become a hole for some public endpoint that happens to redirect at +// the LAN. Rate limiting is per fetcher too, which is the right shape here: +// separate servers are separate hosts. +func WebfetchDoor(limits webfetch.Config) PosterFactory { + return func(cfg ServerConfig) (Poster, error) { + c := limits + c.AllowPrivate = cfg.AllowPrivate + if c.Timeout <= 0 && cfg.Timeout > 0 { + c.Timeout = cfg.Timeout + } + return fetcherPoster{webfetch.New(c)}, nil + } +} + +// fetcherPoster adapts webfetch.Fetcher to Poster. It exists so this package +// does not have to know webfetch's Response type, and so a test can substitute +// a fake without a listener. +type fetcherPoster struct{ f *webfetch.Fetcher } + +func (p fetcherPoster) Post(ctx context.Context, rawURL, contentType string, body []byte, hdr map[string]string) (*PostResponse, error) { + resp, err := p.f.Post(ctx, rawURL, contentType, body, hdr) + if err != nil { + return nil, fmt.Errorf("mcp: post %s: %w", rawURL, err) + } + return &PostResponse{ + Status: resp.Status, + ContentType: resp.ContentType, + Body: resp.Body, + Header: resp.Header, + }, nil +} diff --git a/internal/webfetch/webfetch.go b/internal/webfetch/webfetch.go index 95ab273..5f58050 100644 --- a/internal/webfetch/webfetch.go +++ b/internal/webfetch/webfetch.go @@ -28,6 +28,7 @@ package webfetch import ( + "bytes" "context" "errors" "fmt" @@ -92,6 +93,9 @@ type Response struct { Status int ContentType string Body []byte + // Header — the response headers, one value each (the first). Populated for + // every request; MCP needs Mcp-Session-Id, nothing else reads it. + Header map[string]string } // Fetcher performs guarded GETs. Safe for concurrent use; the per-host rate @@ -159,6 +163,37 @@ func New(cfg Config) *Fetcher { // Get fetches rawURL. The body is capped: a larger response is an error, not a // truncation, because half an XML document is worse than none. func (f *Fetcher) Get(ctx context.Context, rawURL string) (*Response, error) { + return f.do(ctx, http.MethodGet, rawURL, nil, nil) +} + +// Post sends body to rawURL and returns the reply, under exactly the same +// guards as Get: scheme rule, host lists, the dialer's private-address check on +// every hop, the size cap and the per-host rate limit. +// +// It exists for JSON-RPC over HTTP (internal/mcp), which cannot be expressed as +// a GET. That an outbound request now carries a body does not widen the +// address policy one bit — a POST to the LAN is refused for the same reason a +// GET is, unless AllowPrivate was set for that specific fetcher. +// +// hdr is merged over the defaults; a caller may not override User-Agent or +// Accept-Encoding, because identity encoding and an honest UA are part of the +// contract with whatever is on the other end. +func (f *Fetcher) Post(ctx context.Context, rawURL, contentType string, body []byte, hdr map[string]string) (*Response, error) { + if contentType == "" { + contentType = "application/json" + } + if hdr == nil { + hdr = map[string]string{} + } + merged := make(map[string]string, len(hdr)+1) + for k, v := range hdr { + merged[k] = v + } + merged["Content-Type"] = contentType + return f.do(ctx, http.MethodPost, rawURL, body, merged) +} + +func (f *Fetcher) do(ctx context.Context, method, rawURL string, body []byte, hdr map[string]string) (*Response, error) { u, err := url.Parse(strings.TrimSpace(rawURL)) if err != nil { return nil, fmt.Errorf("webfetch: bad url %q: %w", rawURL, err) @@ -170,10 +205,17 @@ func (f *Fetcher) Get(ctx context.Context, rawURL string) (*Response, error) { return nil, err } - req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil) + var rdr io.Reader + if body != nil { + rdr = bytes.NewReader(body) + } + req, err := http.NewRequestWithContext(ctx, method, u.String(), rdr) if err != nil { return nil, err } + for k, v := range hdr { + req.Header.Set(k, v) + } req.Header.Set("User-Agent", f.cfg.UserAgent) req.Header.Set("Accept-Encoding", "identity") @@ -190,22 +232,27 @@ func (f *Fetcher) Get(ctx context.Context, rawURL string) (*Response, error) { } defer resp.Body.Close() - body, err := io.ReadAll(io.LimitReader(resp.Body, f.cfg.MaxBytes+1)) + respBody, err := io.ReadAll(io.LimitReader(resp.Body, f.cfg.MaxBytes+1)) if err != nil { return nil, err } - if int64(len(body)) > f.cfg.MaxBytes { + if int64(len(respBody)) > f.cfg.MaxBytes { return nil, fmt.Errorf("%w (%d bytes)", ErrTooLarge, f.cfg.MaxBytes) } if resp.StatusCode < 200 || resp.StatusCode > 299 { return nil, fmt.Errorf("%w: %d", ErrStatus, resp.StatusCode) } - return &Response{ + out := &Response{ URL: resp.Request.URL.String(), Status: resp.StatusCode, ContentType: resp.Header.Get("Content-Type"), - Body: body, - }, nil + Body: respBody, + Header: map[string]string{}, + } + for k := range resp.Header { + out.Header[k] = resp.Header.Get(k) + } + return out, nil } // checkURL applies the scheme rule and the host lists. The address rule is the diff --git a/internal/webfetch/webfetch_test.go b/internal/webfetch/webfetch_test.go index 667190d..fafdabb 100644 --- a/internal/webfetch/webfetch_test.go +++ b/internal/webfetch/webfetch_test.go @@ -3,6 +3,7 @@ package webfetch import ( "context" "errors" + "io" "net" "net/http" "net/http/httptest" @@ -225,3 +226,74 @@ func TestUserAgentIsSent(t *testing.T) { t.Fatalf("user-agent = %q", ua) } } + +func TestPostSendsBodyAndHeaders(t *testing.T) { + type seen struct { + method, ctype, accept, ua, custom string + body []byte + } + ch := make(chan seen, 1) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + ch <- seen{r.Method, r.Header.Get("Content-Type"), r.Header.Get("Accept"), + r.Header.Get("User-Agent"), r.Header.Get("X-Thing"), b} + w.Header().Set("Mcp-Session-Id", "sess-9") + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + defer srv.Close() + f := testFetcher(t, Config{UserAgent: "Maven/test"}) + resp, err := f.Post(context.Background(), srv.URL, "application/json", + []byte(`{"jsonrpc":"2.0"}`), map[string]string{"Accept": "text/event-stream", "X-Thing": "1"}) + if err != nil { + t.Fatal(err) + } + if string(resp.Body) != `{"ok":true}` { + t.Fatalf("body = %q", resp.Body) + } + if resp.Header["Mcp-Session-Id"] != "sess-9" { + t.Fatalf("response headers not surfaced: %+v", resp.Header) + } + s := <-ch + if s.method != http.MethodPost { + t.Fatalf("method = %s", s.method) + } + if string(s.body) != `{"jsonrpc":"2.0"}` { + t.Fatalf("request body = %q", s.body) + } + if s.ctype != "application/json" { + t.Fatalf("content-type = %q", s.ctype) + } + if s.accept != "text/event-stream" || s.custom != "1" { + t.Fatalf("caller headers dropped: %+v", s) + } + if s.ua != "Maven/test" { + t.Fatalf("user-agent = %q — a caller must not be able to override it", s.ua) + } +} + +// The whole point of routing MCP through webfetch: a POST is guarded exactly +// like a GET. A body does not buy a caller a way onto the LAN. +func TestPostRefusesPrivateAddress(t *testing.T) { + f := New(Config{}) // no AllowPrivate + _, err := f.Post(context.Background(), "http://127.0.0.1:9100/mcp", "application/json", []byte(`{}`), nil) + if !errors.Is(err, ErrPrivate) { + t.Fatalf("error = %v, want ErrPrivate", err) + } +} + +func TestPostRefusesNonHTTPScheme(t *testing.T) { + f := New(Config{}) + if _, err := f.Post(context.Background(), "file:///etc/passwd", "application/json", nil, nil); !errors.Is(err, ErrScheme) { + t.Fatalf("error = %v, want ErrScheme", err) + } +} + +func TestPostObeysDenylist(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + defer srv.Close() + f := testFetcher(t, Config{DenyHosts: []string{"127.0.0.1"}}) + if _, err := f.Post(context.Background(), srv.URL, "application/json", []byte(`{}`), nil); !errors.Is(err, ErrBlocked) { + t.Fatalf("error = %v, want ErrBlocked", err) + } +}