diff --git a/README.md b/README.md index 18cf219..ebb802d 100644 --- a/README.md +++ b/README.md @@ -291,9 +291,26 @@ nothing is inherited from the orcli process, on the same grounds the API key is never read from it. A server is not started until the model calls `mcp_list_resources` or `mcp_read_resource` naming it, and is kept running for reuse by later calls in the same session once it is. Only resource listing -and reading are offered; a server's own tools and prompts are not surfaced, -and a remote server reached over HTTP or SSE rather than launched locally is -not supported today. +and reading are offered; a server's own tools and prompts are not surfaced. + +A server is also reachable remotely over the Streamable HTTP transport +(issue #62) - the one the MCP specification currently recommends for new +integrations; the older HTTP+SSE transport is not supported. Name `url` +instead of `command`, with optional `headers` (a bearer token or any other +credential the remote server itself needs, never read from this process's +own environment, the same rule `env` follows for a local server): + +```json +{"mcp":{"servers":[{"name":"watchgoose","url":"https://watchgoose.example.com/mcp","headers":{"Authorization":"Bearer your-token"}}]}} +``` + +A server names `command` (local) or `url` (remote), never both and never +neither - either is refused by name at configuration read time. A local and +a remote server can be configured side by side; `mcp_list_resources` and +`mcp_read_resource` work identically against either kind once a server is +named, since both speak the same JSON-RPC shape underneath. A remote +server's session (if it assigns one) is torn down with a best-effort DELETE +when the session ends, the same way a local server's subprocess is stopped. Task operations in Notion require the integration token to be shared with the task data source, plus explicit IDs/property names that bind task operations to diff --git a/cmd/orcli/ask.go b/cmd/orcli/ask.go index 21fafdf..0c73aef 100644 --- a/cmd/orcli/ask.go +++ b/cmd/orcli/ask.go @@ -312,7 +312,10 @@ func newToolsetWithTasks(dir, notionToken string, apiarySettings []string, integ func newMCPManager(servers []config.MCPServer) *mcp.Manager { set := make([]mcp.Server, 0, len(servers)) for _, s := range servers { - set = append(set, mcp.Server{Name: s.Name, Command: s.Command, Args: s.Args, Env: s.Env}) + set = append(set, mcp.Server{ + Name: s.Name, Command: s.Command, Args: s.Args, Env: s.Env, + URL: s.URL, Headers: s.Headers, + }) } return mcp.NewManager(set) } diff --git a/cmd/orcli/ask_test.go b/cmd/orcli/ask_test.go index 81a2fa9..03eb9cf 100644 --- a/cmd/orcli/ask_test.go +++ b/cmd/orcli/ask_test.go @@ -8,6 +8,7 @@ import ( "strings" "testing" + "github.com/glenjbarber/orcli/internal/config" "github.com/glenjbarber/orcli/internal/openrouter" "github.com/glenjbarber/orcli/internal/tools" "github.com/glenjbarber/orcli/internal/tui" @@ -695,3 +696,26 @@ func TestSilentTurnStillConfirmsTheModel(t *testing.T) { t.Errorf("confirmed(%q, %v), want (\"some/model\", true)", gotModel, gotSilent) } } + +// TestNewMCPManagerPassesTheRemoteServerFieldsThrough checks the adapter +// forwards URL/Headers (issue #62), not only the local Command/Args/Env +// fields config.MCPServer already carried - configuredServers() (used in +// every tool's own description) names every configured server regardless of +// kind, which is enough to prove the remote one actually reached the +// Manager rather than being dropped on the way. +func TestNewMCPManagerPassesTheRemoteServerFieldsThrough(t *testing.T) { + m := newMCPManager([]config.MCPServer{ + {Name: "local-files", Command: "npx", Args: []string{"-y", "server-filesystem"}}, + {Name: "watchgoose", URL: "https://watchgoose.example.com/mcp", Headers: map[string]string{"Authorization": "Bearer t"}}, + }) + defer m.Close() + + set := m.ToolSet() + if len(set) == 0 { + t.Fatal("ToolSet() returned no tools") + } + desc := set[0].Describe().Function.Description + if !strings.Contains(desc, "local-files") || !strings.Contains(desc, "watchgoose") { + t.Fatalf("tool description = %q, want it to name both configured servers", desc) + } +} diff --git a/internal/config/load_test.go b/internal/config/load_test.go index ed9027a..8a91c55 100644 --- a/internal/config/load_test.go +++ b/internal/config/load_test.go @@ -43,8 +43,11 @@ func TestExampleConfigLoadsThroughTheRealParser(t *testing.T) { if err != nil { t.Fatalf("MCPServers() = %v, want nil", err) } - if len(servers) != 1 || servers[0].Name != "example" { - t.Errorf("MCPServers() = %+v, want one server named %q", servers, "example") + if len(servers) != 2 || servers[0].Name != "example" { + t.Errorf("MCPServers() = %+v, want two servers, the first named %q", servers, "example") + } + if servers[1].Name != "remote-example" || servers[1].URL == "" || len(servers[1].Headers) == 0 { + t.Errorf("MCPServers()[1] = %+v, want the remote example with a URL and headers", servers[1]) } } diff --git a/internal/config/mcp.go b/internal/config/mcp.go index 62cb1e5..1b9f1f1 100644 --- a/internal/config/mcp.go +++ b/internal/config/mcp.go @@ -25,27 +25,39 @@ func (m MCP) MarshalJSON() ([]byte, error) { return m.raw, nil } -// MCPServer is one Model Context Protocol server the reader has configured, -// launched as a subprocess and spoken to over its standard input and output. -// -// Only this shape - a local command, launched over stdio - is read today. A -// remote server reached over HTTP or SSE is not a configuration this client -// understands yet. +// MCPServer is one Model Context Protocol server the reader has configured - +// either a local command launched as a subprocess and spoken to over its +// standard input and output, or a remote server reached over HTTP using the +// Streamable HTTP transport (the one the specification currently +// recommends; the older HTTP+SSE transport is not supported). The two are +// mutually exclusive: MCPServers refuses a server naming both a command and +// a URL, or neither. type MCPServer struct { // Name is how the model names this server when it calls a tool. Name string // Command is the program to launch, resolved by the search path. + // Empty for a remote server. Command string - // Args are the arguments the program is started with. + // Args are the arguments the program is started with. Local only. Args []string // Env is the subprocess's environment, as KEY=VALUE pairs, in addition to // PATH. It is never read from this process's own environment: a server // that needs a credential gets it because the reader wrote it here, the - // same way every other credential in this file is handled. + // same way every other credential in this file is handled. Local only. Env []string + + // URL is the remote server's own Streamable HTTP endpoint. Empty for a + // local server. + URL string + + // Headers are sent with every request to a remote server - this + // transport's own equivalent of Env, for a bearer token or any other + // credential the server itself needs, written here rather than read + // from this process's own environment. Remote only. + Headers map[string]string } // MCPServers returns the MCP servers configured under the mcp block. @@ -68,6 +80,8 @@ func (c Config) MCPServers() ([]MCPServer, error) { Command string `json:"command"` Args []string `json:"args"` Env map[string]string `json:"env"` + URL string `json:"url"` + Headers map[string]string `json:"headers"` } `json:"servers"` } if err := json.Unmarshal(c.MCP.raw, &block); err != nil { @@ -79,37 +93,49 @@ func (c Config) MCPServers() ([]MCPServer, error) { for _, s := range block.Servers { name := strings.TrimSpace(s.Name) command := strings.TrimSpace(s.Command) + url := strings.TrimSpace(s.URL) if name == "" { return nil, fmt.Errorf("config: %s: a server with no name was given", mcpKey) } - if command == "" { - return nil, fmt.Errorf("config: %s: server %q has no command", mcpKey, name) + if command == "" && url == "" { + return nil, fmt.Errorf("config: %s: server %q has neither a command nor a url", mcpKey, name) + } + if command != "" && url != "" { + return nil, fmt.Errorf("config: %s: server %q names both a command and a url; a server is local (command) or remote (url), never both", mcpKey, name) } if _, exists := seen[name]; exists { return nil, fmt.Errorf("config: %s: server %q is listed more than once", mcpKey, name) } seen[name] = struct{}{} - keys := make([]string, 0, len(s.Env)) - for k := range s.Env { - keys = append(keys, k) - } - sort.Strings(keys) - env := make([]string, 0, len(keys)) - for _, k := range keys { - env = append(env, k+"="+s.Env[k]) - } - servers = append(servers, MCPServer{ Name: name, Command: command, Args: append([]string(nil), s.Args...), - Env: env, + Env: sortedKeyValues(s.Env), + URL: url, + Headers: s.Headers, }) } return servers, nil } +// sortedKeyValues renders a map as KEY=VALUE strings, sorted by key, so the +// order is stable across reads of the same file rather than following Go's +// own randomized map iteration. +func sortedKeyValues(m map[string]string) []string { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + sort.Strings(keys) + out := make([]string, 0, len(keys)) + for _, k := range keys { + out = append(out, k+"="+m[k]) + } + return out +} + func readMCP(raw json.RawMessage) (*MCP, error) { if isNull(raw) || len(bytes.TrimSpace(raw)) == 0 { return nil, nil diff --git a/internal/config/mcp_test.go b/internal/config/mcp_test.go index 9576cb0..361ba11 100644 --- a/internal/config/mcp_test.go +++ b/internal/config/mcp_test.go @@ -41,6 +41,65 @@ func TestMCPServersAreReadAndUnknownMembersSurvive(t *testing.T) { } } +// TestMCPServersReadsARemoteServer covers a server configured with a url +// instead of a command - the Streamable HTTP shape this client now supports +// (issue #62). +func TestMCPServersReadsARemoteServer(t *testing.T) { + h := home(t) + writeConfigAt(t, h, `{"api_key":"openrouter","mcp":{"servers":[{"name":"watchgoose","url":"https://watchgoose.example.com/mcp","headers":{"Authorization":"Bearer secret-token"}}]}}`) + + cfg, err := Load() + if err != nil { + t.Fatalf("Load: %v", err) + } + servers, err := cfg.MCPServers() + if err != nil { + t.Fatalf("MCPServers: %v", err) + } + if len(servers) != 1 { + t.Fatalf("MCPServers = %#v, want one server", servers) + } + got := servers[0] + if got.Name != "watchgoose" || got.URL != "https://watchgoose.example.com/mcp" { + t.Fatalf("MCPServers[0] = %#v", got) + } + if got.Command != "" { + t.Errorf("a remote server carried a command: %q", got.Command) + } + if got.Headers["Authorization"] != "Bearer secret-token" { + t.Errorf("MCPServers[0].Headers = %#v", got.Headers) + } +} + +// TestMCPServersRejectBothCommandAndURL and +// TestMCPServersRejectNeitherCommandNorURL guard the mutual-exclusion rule: +// a server is local or remote, never both and never neither. +func TestMCPServersRejectBothCommandAndURL(t *testing.T) { + h := home(t) + writeConfigAt(t, h, `{"api_key":"openrouter","mcp":{"servers":[{"name":"files","command":"npx","url":"https://example.com/mcp"}]}}`) + cfg, err := Load() + if err != nil { + t.Fatalf("Load: %v", err) + } + _, err = cfg.MCPServers() + if err == nil || !strings.Contains(err.Error(), "both a command and a url") { + t.Fatalf("MCPServers() = %v, want a refusal naming both a command and a url", err) + } +} + +func TestMCPServersRejectNeitherCommandNorURL(t *testing.T) { + h := home(t) + writeConfigAt(t, h, `{"api_key":"openrouter","mcp":{"servers":[{"name":"files"}]}}`) + cfg, err := Load() + if err != nil { + t.Fatalf("Load: %v", err) + } + _, err = cfg.MCPServers() + if err == nil || !strings.Contains(err.Error(), "neither a command nor a url") { + t.Fatalf("MCPServers() = %v, want a refusal naming neither", err) + } +} + func TestMCPServersMissingBlockOrEmpty(t *testing.T) { for _, body := range []string{ `{"api_key":"openrouter"}`, diff --git a/internal/mcp/http_transport.go b/internal/mcp/http_transport.go new file mode 100644 index 0000000..c3d7156 --- /dev/null +++ b/internal/mcp/http_transport.go @@ -0,0 +1,252 @@ +package mcp + +import ( + "bufio" + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "sync" + "sync/atomic" +) + +// mcpSessionHeader is the header a Streamable HTTP server assigns a session +// ID under in its initialize response, and the header every later request in +// the same session carries it back under (MCP spec 2025-03-26). +const mcpSessionHeader = "Mcp-Session-Id" + +// httpTransport is one remote MCP server reached over the Streamable HTTP +// transport: every JSON-RPC message is POSTed, as its own HTTP request, to +// one configured URL. +// +// What is implemented: POSTing a request and reading back either a single +// `application/json` response body or a single-event `text/event-stream` +// response (a server may answer either way per the spec, and both are +// read here), session continuity via Mcp-Session-Id once a server assigns +// one, and best-effort session termination (DELETE) on Close. +// +// What is deliberately not implemented: a standalone server-initiated SSE +// stream opened with a bare GET to the same URL, used by a server to push a +// notification or request the client did not ask for. This client does not +// act on an unsolicited server message over stdio either (stdioTransport's +// own readLoop skips a line with no numeric id for the same reason), so +// there being no open GET stream to deliver one over is not a narrowing of +// behaviour this client already has - it is the same behaviour reached a +// second way. Resumable streams (Last-Event-ID) are not implemented either, +// since there is no standing stream to resume without that GET. +type httpTransport struct { + url string + headers map[string]string + client *http.Client + + mu sync.Mutex + sessionID string + + nextID int64 +} + +func newHTTPTransport(url string, headers map[string]string) *httpTransport { + return &httpTransport{url: url, headers: headers, client: &http.Client{Timeout: callTimeout}} +} + +// session returns the session ID this transport has been assigned, if any. +func (t *httpTransport) session() string { + t.mu.Lock() + defer t.mu.Unlock() + return t.sessionID +} + +// setSession records a session ID a response assigned. A server is not +// required to assign one (the spec allows a sessionless server), so an empty +// value from a response changes nothing - it is never used to clear a +// session this transport already has. +func (t *httpTransport) setSession(id string) { + if id == "" { + return + } + t.mu.Lock() + t.sessionID = id + t.mu.Unlock() +} + +// post sends one JSON-RPC message (request or notification) and returns the +// raw HTTP response, with every header and the session ID this transport +// has on hand attached. +func (t *httpTransport) post(body []byte) (*http.Response, error) { + req, err := http.NewRequest(http.MethodPost, t.url, bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf("mcp: build request: %w", 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 id := t.session(); id != "" { + req.Header.Set(mcpSessionHeader, id) + } + + resp, err := t.client.Do(req) + if err != nil { + return nil, fmt.Errorf("mcp: %s: %w", t.url, err) + } + t.setSession(resp.Header.Get(mcpSessionHeader)) + return resp, nil +} + +// call sends a request and returns its result, or an error naming what went +// wrong - a non-2xx status, a body that is neither JSON nor a parseable +// single-event SSE stream, or a JSON-RPC error object in the response. +func (t *httpTransport) call(method string, params any) (json.RawMessage, error) { + id := atomic.AddInt64(&t.nextID, 1) + req := rpcRequest{JSONRPC: "2.0", ID: id, Method: method, Params: params} + data, err := json.Marshal(req) + if err != nil { + return nil, fmt.Errorf("mcp: encode %s: %w", method, err) + } + + resp, err := t.post(data) + if err != nil { + return nil, fmt.Errorf("mcp: %s: %w", method, err) + } + defer resp.Body.Close() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + detail, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return nil, fmt.Errorf("mcp: %s: %s returned %s: %s", method, t.url, resp.Status, strings.TrimSpace(string(detail))) + } + + rpcResp, err := decodeRPCResponse(resp) + if err != nil { + return nil, fmt.Errorf("mcp: %s: %w", method, err) + } + if rpcResp.Error != nil { + return nil, fmt.Errorf("mcp: %s: %s (code %d)", method, rpcResp.Error.Message, rpcResp.Error.Code) + } + return rpcResp.Result, nil +} + +// notify sends a JSON-RPC message with no reply expected. Per the spec a +// server answers a notification-only POST with 202 Accepted and an empty +// body; any 2xx is accepted here without trying to parse a body that may not +// exist. +func (t *httpTransport) notify(method string, params any) error { + note := rpcNotification{JSONRPC: "2.0", Method: method, Params: params} + data, err := json.Marshal(note) + if err != nil { + return fmt.Errorf("mcp: encode %s: %w", method, err) + } + + resp, err := t.post(data) + if err != nil { + return fmt.Errorf("mcp: %s: %w", method, err) + } + defer io.Copy(io.Discard, resp.Body) + defer resp.Body.Close() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("mcp: %s: %s returned %s", method, t.url, resp.Status) + } + return nil +} + +// decodeRPCResponse reads a POST response body as either a bare JSON object +// or a single-event text/event-stream body, whichever Content-Type the +// server actually sent - the spec permits either for a request that expects +// a reply. +func decodeRPCResponse(resp *http.Response) (rpcResponse, error) { + contentType := resp.Header.Get("Content-Type") + switch { + case strings.Contains(contentType, "text/event-stream"): + return decodeSSEResponse(resp.Body) + default: + // application/json is the expected case; anything else is still + // handed to the JSON decoder rather than refused up front, since a + // server that omitted or mislabelled the header but sent a legal + // JSON body should still work rather than being refused on a + // technicality this client can see past. + var out rpcResponse + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return rpcResponse{}, fmt.Errorf("decode response body (content-type %q): %w", contentType, err) + } + return out, nil + } +} + +// decodeSSEResponse reads one Server-Sent Events stream until it finds a +// `data:` field, parses that as the JSON-RPC response, and stops - a +// Streamable HTTP POST response is a single reply framed as one SSE event, +// not an open-ended stream of many, so reading past the first complete event +// is reading past what this request's own response ever contains. +// +// An SSE event can carry its data split across multiple `data:` lines +// (joined with newlines per the SSE spec) and may carry `event:`/`id:` +// fields this client does not need and skips. The event ends at the first +// blank line, which is this function's own stopping point. +func decodeSSEResponse(r io.Reader) (rpcResponse, error) { + scanner := bufio.NewScanner(r) + scanner.Buffer(make([]byte, 0, 64*1024), 16<<20) + + var data strings.Builder + sawData := false + for scanner.Scan() { + line := scanner.Text() + if line == "" { + if sawData { + break + } + continue + } + field, value, _ := strings.Cut(line, ":") + if field != "data" { + continue + } + if sawData { + data.WriteByte('\n') + } + data.WriteString(strings.TrimPrefix(value, " ")) + sawData = true + } + if err := scanner.Err(); err != nil { + return rpcResponse{}, fmt.Errorf("read event stream: %w", err) + } + if !sawData { + return rpcResponse{}, fmt.Errorf("event stream carried no data field") + } + + var out rpcResponse + if err := json.Unmarshal([]byte(data.String()), &out); err != nil { + return rpcResponse{}, fmt.Errorf("decode event stream data: %w", err) + } + return out, nil +} + +// Close ends this transport's session, if the server assigned one. Per the +// spec a client SHOULD send DELETE to explicitly terminate a session; this +// is best-effort, since there is nothing useful to do locally if a remote +// server does not acknowledge it, and failing to tear down a remote session +// is not a reason to fail the local Close call the rest of this package +// depends on succeeding. +func (t *httpTransport) Close() error { + id := t.session() + if id == "" { + return nil + } + req, err := http.NewRequest(http.MethodDelete, t.url, nil) + if err != nil { + return nil + } + req.Header.Set(mcpSessionHeader, id) + for k, v := range t.headers { + req.Header.Set(k, v) + } + resp, err := t.client.Do(req) + if err != nil { + return nil + } + defer resp.Body.Close() + io.Copy(io.Discard, resp.Body) + return nil +} diff --git a/internal/mcp/http_transport_test.go b/internal/mcp/http_transport_test.go new file mode 100644 index 0000000..ee515bd --- /dev/null +++ b/internal/mcp/http_transport_test.go @@ -0,0 +1,272 @@ +package mcp + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" +) + +// fakeHTTPServer is a minimal Streamable HTTP MCP server: it answers +// initialize, resources/list and resources/read, assigns a session ID on +// initialize, and records every request it saw so a test can assert on +// headers and session continuity. +type fakeHTTPServer struct { + mu sync.Mutex + requests []*http.Request + sseMode bool + + sessionID string + deleteCalled int32 +} + +func newFakeHTTPServer(sseMode bool) *fakeHTTPServer { + return &fakeHTTPServer{sseMode: sseMode, sessionID: "session-abc-123"} +} + +func (f *fakeHTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + // Headers are read now, before the body is consumed, and the request + // itself is cloned-by-reference for later inspection - fine here since + // nothing in this test server mutates it afterward. + f.requests = append(f.requests, r) + f.mu.Unlock() + + if r.Method == http.MethodDelete { + atomic.AddInt32(&f.deleteCalled, 1) + w.WriteHeader(http.StatusNoContent) + return + } + + var req rpcRequest + body, _ := io.ReadAll(r.Body) + if err := json.Unmarshal(body, &req); err != nil { + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + // A notification (no reply expected) is identified here the same way a + // real server would: no "id" member at all. rpcRequest.ID is int64 and + // decodes to 0 either way, so the raw body is checked instead. + if !strings.Contains(string(body), `"id"`) { + w.WriteHeader(http.StatusAccepted) + return + } + + var result any + switch req.Method { + case "initialize": + w.Header().Set(mcpSessionHeader, f.sessionID) + result = map[string]any{"protocolVersion": protocolVersion} + case "resources/list": + result = listResourcesResult{Resources: []Resource{{URI: "mem://a", Name: "a"}}} + case "resources/read": + result = readResourceResult{Contents: []resourceContent{{URI: "mem://a", Text: "hello from remote"}}} + default: + http.Error(w, "unknown method", http.StatusNotFound) + return + } + + resultBytes, _ := json.Marshal(result) + resp := rpcResponse{JSONRPC: "2.0", ID: json.RawMessage(fmt.Sprintf("%d", req.ID)), Result: resultBytes} + respBytes, _ := json.Marshal(resp) + + if f.sseMode { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + fmt.Fprintf(w, "event: message\ndata: %s\n\n", respBytes) + return + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write(respBytes) +} + +func (f *fakeHTTPServer) lastRequest() *http.Request { + f.mu.Lock() + defer f.mu.Unlock() + return f.requests[len(f.requests)-1] +} + +func TestHTTPTransport_JSONResponseMode(t *testing.T) { + fake := newFakeHTTPServer(false) + ts := httptest.NewServer(fake) + defer ts.Close() + + tr := newHTTPTransport(ts.URL, map[string]string{"Authorization": "Bearer test-token"}) + raw, err := tr.call("resources/list", map[string]any{}) + if err != nil { + t.Fatalf("call: %v", err) + } + var result listResourcesResult + if err := json.Unmarshal(raw, &result); err != nil { + t.Fatalf("decode: %v", err) + } + if len(result.Resources) != 1 || result.Resources[0].URI != "mem://a" { + t.Fatalf("result = %+v", result) + } + if got := fake.lastRequest().Header.Get("Authorization"); got != "Bearer test-token" { + t.Errorf("Authorization header = %q", got) + } +} + +func TestHTTPTransport_SSEResponseMode(t *testing.T) { + fake := newFakeHTTPServer(true) + ts := httptest.NewServer(fake) + defer ts.Close() + + tr := newHTTPTransport(ts.URL, nil) + raw, err := tr.call("resources/read", map[string]any{"uri": "mem://a"}) + if err != nil { + t.Fatalf("call: %v", err) + } + var result readResourceResult + if err := json.Unmarshal(raw, &result); err != nil { + t.Fatalf("decode: %v", err) + } + if len(result.Contents) != 1 || result.Contents[0].Text != "hello from remote" { + t.Fatalf("result = %+v", result) + } +} + +func TestHTTPTransport_SessionIDPropagatesAfterInitialize(t *testing.T) { + fake := newFakeHTTPServer(false) + ts := httptest.NewServer(fake) + defer ts.Close() + + tr := newHTTPTransport(ts.URL, nil) + if _, err := tr.call("initialize", map[string]any{}); err != nil { + t.Fatalf("initialize: %v", err) + } + if tr.session() != "session-abc-123" { + t.Fatalf("session() = %q, want the server-assigned id", tr.session()) + } + if _, err := tr.call("resources/list", map[string]any{}); err != nil { + t.Fatalf("resources/list: %v", err) + } + if got := fake.lastRequest().Header.Get(mcpSessionHeader); got != "session-abc-123" { + t.Errorf("the second request did not carry the session id, got %q", got) + } +} + +func TestHTTPTransport_NotifyAcceptsA202WithNoBody(t *testing.T) { + fake := newFakeHTTPServer(false) + ts := httptest.NewServer(fake) + defer ts.Close() + + tr := newHTTPTransport(ts.URL, nil) + if err := tr.notify("notifications/initialized", map[string]any{}); err != nil { + t.Fatalf("notify: %v", err) + } +} + +func TestHTTPTransport_NonSuccessStatusIsReported(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "nope", http.StatusForbidden) + })) + defer ts.Close() + + tr := newHTTPTransport(ts.URL, nil) + _, err := tr.call("resources/list", map[string]any{}) + if err == nil || !strings.Contains(err.Error(), "403") { + t.Fatalf("call() = %v, want an error naming the 403 status", err) + } +} + +func TestHTTPTransport_RPCErrorObjectIsReported(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req rpcRequest + body, _ := io.ReadAll(r.Body) + json.Unmarshal(body, &req) + resp := rpcResponse{JSONRPC: "2.0", ID: json.RawMessage(fmt.Sprintf("%d", req.ID)), Error: &rpcError{Code: -32601, Message: "method not found"}} + respBytes, _ := json.Marshal(resp) + w.Header().Set("Content-Type", "application/json") + w.Write(respBytes) + })) + defer ts.Close() + + tr := newHTTPTransport(ts.URL, nil) + _, err := tr.call("resources/list", map[string]any{}) + if err == nil || !strings.Contains(err.Error(), "method not found") { + t.Fatalf("call() = %v, want the JSON-RPC error message", err) + } +} + +func TestHTTPTransport_CloseSendsDeleteOnlyWhenASessionExists(t *testing.T) { + fake := newFakeHTTPServer(false) + ts := httptest.NewServer(fake) + defer ts.Close() + + tr := newHTTPTransport(ts.URL, nil) + if err := tr.Close(); err != nil { + t.Fatalf("Close() with no session established: %v", err) + } + if atomic.LoadInt32(&fake.deleteCalled) != 0 { + t.Error("Close sent DELETE despite no session ever being assigned") + } + + if _, err := tr.call("initialize", map[string]any{}); err != nil { + t.Fatalf("initialize: %v", err) + } + if err := tr.Close(); err != nil { + t.Fatalf("Close() after a session was assigned: %v", err) + } + if atomic.LoadInt32(&fake.deleteCalled) != 1 { + t.Errorf("deleteCalled = %d, want 1", fake.deleteCalled) + } +} + +// TestRemoteServerEndToEnd proves a Server naming a URL works through the +// Manager exactly the way a local one does: initialize, list, and read, all +// through client's own transport-agnostic protocol methods. +func TestRemoteServerEndToEnd(t *testing.T) { + fake := newFakeHTTPServer(false) + ts := httptest.NewServer(fake) + defer ts.Close() + + m := NewManager([]Server{{Name: "remote", URL: ts.URL}}) + defer m.Close() + + c, err := m.get("remote") + if err != nil { + t.Fatalf("get: %v", err) + } + resources, err := c.listResources() + if err != nil { + t.Fatalf("listResources: %v", err) + } + if len(resources) != 1 || resources[0].URI != "mem://a" { + t.Fatalf("listResources() = %+v", resources) + } + text, err := c.readResource("mem://a") + if err != nil { + t.Fatalf("readResource: %v", err) + } + if text != "hello from remote" { + t.Fatalf("readResource() = %q", text) + } +} + +func TestDecodeSSEResponse_MultiLineData(t *testing.T) { + raw := "event: message\ndata: {\"jsonrpc\":\"2.0\",\"id\":1,\n" + + "data: \"result\":{}}\n\n" + resp, err := decodeSSEResponse(strings.NewReader(raw)) + if err != nil { + t.Fatalf("decodeSSEResponse: %v", err) + } + if resp.JSONRPC != "2.0" { + t.Fatalf("resp = %+v", resp) + } +} + +func TestDecodeSSEResponse_NoDataFieldIsAnError(t *testing.T) { + _, err := decodeSSEResponse(strings.NewReader("event: ping\n\n")) + if err == nil { + t.Fatal("expected an error for a stream with no data field") + } +} diff --git a/internal/mcp/mcp.go b/internal/mcp/mcp.go index 17b06aa..3d197f4 100644 --- a/internal/mcp/mcp.go +++ b/internal/mcp/mcp.go @@ -1,38 +1,51 @@ // Package mcp provides model-callable tools backed by Model Context Protocol -// servers the reader has configured, each launched as a subprocess and spoken -// to over its standard input and output. +// servers the reader has configured. A server is reached one of two ways: a +// local command launched as a subprocess and spoken to over its standard +// input and output, or a remote server reached over HTTP using the +// Streamable HTTP transport (MCP spec 2025-03-26) - the transport the +// specification currently recommends for new integrations. The older +// HTTP+SSE transport is legacy and is not implemented; a server that only +// offers it is not reachable from this client. // // Only resource listing and resource reading are offered today. A server's -// tools and prompts are not surfaced, and a remote server reached over HTTP -// or SSE rather than launched as a local subprocess is not supported; both -// are tracked as follow-ups rather than built here. +// tools and prompts are not surfaced. // // # Transport // -// The wire format is JSON-RPC 2.0, one message per line, matching the +// Both transports speak the same JSON-RPC 2.0 request/notification/response +// shape; only how a message physically travels differs, which is why +// client's own protocol methods (initialize, listResources, readResource) +// are written once against the transport interface below and never need to +// know which kind of server they are talking to. +// +// The stdio transport is one message per line on stdin/stdout, matching the // protocol's stdio transport. A subprocess is given PATH plus whatever the // reader named under the server's own env block in the configuration file, // and nothing else: the same rule internal/tools states for the shell and // git tools, so a server this package starts cannot inherit a credential the // reader did not explicitly hand it. // +// The HTTP transport POSTs one JSON-RPC message per call to the server's +// configured URL and reads back either a single `application/json` response +// body or a single-event `text/event-stream` response, whichever the server +// sends (see http_transport.go's own doc comment for what is and is not +// implemented - notably, a standalone server-initiated SSE stream opened +// with GET is not, since this client does not act on unsolicited server +// notifications over stdio either). +// // # Lazy, and kept once started // // A server named in the configuration file is not started until the model // actually calls a tool naming it. A session that never asks a configured -// server for anything never pays the cost of a subprocess for it. Once -// started, a server is kept running and reused by later calls in the same -// session rather than relaunched each time, since the handshake a server -// requires before its first real request is not free either. +// server for anything never pays the cost of starting it. Once started, a +// server is kept and reused by later calls in the same session rather than +// relaunched or re-dialed each time, since the handshake a server requires +// before its first real request is not free either. package mcp import ( - "bufio" "encoding/json" "fmt" - "io" - "os" - "os/exec" "sort" "strings" "sync" @@ -43,8 +56,9 @@ import ( // callTimeout bounds how long a single JSON-RPC round trip may take before it // is reported as timed out rather than left to hang the turn that asked for -// it. A local subprocess that has not answered within this window is treated -// as broken, and the next call to the same server starts a fresh one. +// it. A local subprocess or a remote server that has not answered within this +// window is treated as broken, and the next call to the same server starts a +// fresh one. // // It is a variable, not a constant, only so a test can shrink it rather than // run for the full 30 seconds to prove a timeout is reported; production @@ -60,8 +74,13 @@ const protocolVersion = "2024-11-05" // one tool call into an unbounded loop. const maxResourcePages = 64 -// Server names one MCP server the reader has configured, by the command and -// arguments that launch it. +// Server names one MCP server the reader has configured. +// +// A local server names Command (and, optionally, Args/Env); a remote one +// names URL (and, optionally, Headers) instead. The two are mutually +// exclusive - internal/config.MCPServers refuses a server naming both or +// neither before this package ever sees it, so start treats a non-empty URL +// as the only signal it needs to pick the HTTP transport. type Server struct { // Name is how the model names this server when it calls a tool. It is // the key the reader chose in the configuration file, not something @@ -69,22 +88,43 @@ type Server struct { Name string // Command is the program to run, resolved by the search path the way - // internal/tools resolves the shell tool's programs. + // internal/tools resolves the shell tool's programs. Empty for a + // remote server. Command string - // Args are the arguments the program is started with. + // Args are the arguments the program is started with. Local only. Args []string // Env is the subprocess's environment, as KEY=VALUE pairs, in addition // to PATH. It is never read from this process's own environment: a // server that needs a credential gets it because the reader wrote it // into the server's own env block, the same way every other credential - // in the configuration file is handled. + // in the configuration file is handled. Local only. Env []string + + // URL is the remote server's own Streamable HTTP endpoint - one URL + // every JSON-RPC message is POSTed to. Empty for a local server. + URL string + + // Headers are sent with every request to a remote server (for example + // Authorization), the HTTP transport's own equivalent of Env: a + // credential a remote server needs is here because the reader wrote it + // here, never read from this process's own environment. Remote only. + Headers map[string]string +} + +// transport is the seam client's protocol methods are written against, so +// initialize/listResources/readResource never need to know whether they are +// talking to a subprocess or a remote endpoint. +type transport interface { + call(method string, params any) (json.RawMessage, error) + notify(method string, params any) error + Close() error } // Manager holds the MCP servers a session was configured with, and starts -// each one's subprocess lazily, on the first call that names it. +// each one's subprocess or opens its HTTP session lazily, on the first call +// that names it. type Manager struct { mu sync.Mutex servers map[string]Server @@ -146,8 +186,8 @@ func (m *Manager) configuredServers() string { return "configured servers: " + strings.Join(names, ", ") } -// get returns the running client for a configured server, starting and -// initializing it on first use. +// get returns the running client for a configured server, starting (local) +// or dialing (remote) and initializing it on first use. func (m *Manager) get(name string) (*client, error) { m.mu.Lock() defer m.mu.Unlock() @@ -175,7 +215,8 @@ func (m *Manager) get(name string) (*client, error) { } // drop closes and discards a server's client, so the next call to it starts -// a fresh subprocess rather than reusing one a prior call found broken. +// a fresh subprocess or HTTP session rather than reusing one a prior call +// found broken. func (m *Manager) drop(name string) { m.mu.Lock() c, ok := m.clients[name] @@ -205,9 +246,9 @@ type rpcNotification struct { } // rpcResponse is what a server sends back. ID is read as raw JSON because an -// MCP server's own notification to the client arrives on the same stream with -// no id at all, and this client tells the two apart by whether ID decodes to -// a number rather than by a separate shape. +// MCP server's own notification to the client can arrive on the same stream +// with no id at all, and this client tells the two apart by whether ID +// decodes to a number rather than by a separate shape. type rpcResponse struct { JSONRPC string `json:"jsonrpc"` ID json.RawMessage `json:"id"` @@ -221,187 +262,46 @@ type rpcError struct { Message string `json:"message"` } -// client is one running MCP server, spoken to over newline-delimited -// JSON-RPC on its stdin and stdout. -// -// A single background goroutine owns the read side and demultiplexes -// responses to the call that is waiting for each one by ID, which is what -// lets a call time out without corrupting a read another call still needs: -// the reader never stops, it only stops being waited on. +// client is one running MCP server - local or remote - spoken to through its +// own transport. Every protocol-level method below (initialize, +// listResources, readResource) is written once against transport and works +// unchanged for either kind of server. type client struct { - cmd *exec.Cmd - stdin io.WriteCloser - - mu sync.Mutex - nextID int64 - pending map[int64]chan rpcResponse - readErr error + t transport } -// start launches a server's subprocess and begins reading its output. It -// does not perform the initialize handshake; the caller does that once the -// client is constructed, so a failed handshake can close what start opened. +// start opens a server's transport - launching a subprocess for a local +// server, or preparing an HTTP session for a remote one - based on which of +// Command/URL the Server names. It does not perform the initialize +// handshake; the caller does that once the client is constructed, so a +// failed handshake can close what start opened. func start(s Server) (*client, error) { - cmd := exec.Command(s.Command, s.Args...) - cmd.Env = append([]string{"PATH=" + os.Getenv("PATH")}, s.Env...) - - stdin, err := cmd.StdinPipe() - if err != nil { - return nil, fmt.Errorf("mcp: open stdin: %w", err) + if s.URL != "" { + return &client{t: newHTTPTransport(s.URL, s.Headers)}, nil } - stdout, err := cmd.StdoutPipe() + t, err := newStdioTransport(s) if err != nil { - return nil, fmt.Errorf("mcp: open stdout: %w", err) - } - // Stderr is discarded rather than captured: a server's diagnostic chatter - // is not a resource and is not shown to the model, and capturing it - // without a bound would be one more place a hung or chatty server could - // grow without limit. - cmd.Stderr = io.Discard - - if err := cmd.Start(); err != nil { - return nil, fmt.Errorf("mcp: start %s: %w", s.Command, err) - } - - c := &client{cmd: cmd, stdin: stdin, pending: make(map[int64]chan rpcResponse)} - go c.readLoop(stdout) - return c, nil -} - -// readLoop reads one JSON-RPC message per line for the life of the -// subprocess, and delivers each to whichever call is waiting for its ID. A -// line this client cannot parse, or one carrying no numeric ID - a -// notification a server sent unasked - is skipped rather than treated as a -// fault: this client has nothing it is waiting to do with either. -func (c *client) readLoop(r io.Reader) { - scanner := bufio.NewScanner(r) - scanner.Buffer(make([]byte, 0, 64*1024), 16<<20) - - for scanner.Scan() { - line := strings.TrimSpace(scanner.Text()) - if line == "" { - continue - } - var resp rpcResponse - if json.Unmarshal([]byte(line), &resp) != nil { - continue - } - var id int64 - if json.Unmarshal(resp.ID, &id) != nil { - continue - } - - c.mu.Lock() - ch, ok := c.pending[id] - if ok { - delete(c.pending, id) - } - c.mu.Unlock() - - if ok { - ch <- resp - } - } - - err := fmt.Errorf("mcp: the server's output ended") - if scanner.Err() != nil { - err = fmt.Errorf("mcp: read: %w", scanner.Err()) - } - - c.mu.Lock() - c.readErr = err - pending := c.pending - c.pending = make(map[int64]chan rpcResponse) - c.mu.Unlock() - - for _, ch := range pending { - close(ch) + return nil, err } + return &client{t: t}, nil } // call sends a request and waits for its matching response, or for -// callTimeout to pass. -// -// A call that times out drops its own slot so the (possibly still arriving) -// response is discarded by readLoop rather than delivered to no one, but it -// does not stop the subprocess: a slow answer is not evidence the server is -// broken, and the next call gets its own fresh wait. +// callTimeout to pass. Delegates to the transport; see stdioTransport.call +// and httpTransport.call for how each kind actually waits. func (c *client) call(method string, params any) (json.RawMessage, error) { - c.mu.Lock() - c.nextID++ - id := c.nextID - ch := make(chan rpcResponse, 1) - c.pending[id] = ch - c.mu.Unlock() - - req := rpcRequest{JSONRPC: "2.0", ID: id, Method: method, Params: params} - data, err := json.Marshal(req) - if err != nil { - c.forget(id) - return nil, fmt.Errorf("mcp: encode %s: %w", method, err) - } - data = append(data, '\n') - - if _, err := c.stdin.Write(data); err != nil { - c.forget(id) - return nil, fmt.Errorf("mcp: write %s: %w", method, err) - } - - select { - case resp, ok := <-ch: - if !ok { - c.mu.Lock() - readErr := c.readErr - c.mu.Unlock() - return nil, fmt.Errorf("mcp: %s: %w", method, readErr) - } - if resp.Error != nil { - return nil, fmt.Errorf("mcp: %s: %s (code %d)", method, resp.Error.Message, resp.Error.Code) - } - return resp.Result, nil - - case <-time.After(callTimeout): - c.forget(id) - return nil, fmt.Errorf("mcp: %s timed out after %s", method, callTimeout) - } -} - -// forget removes a call's slot without waiting for it further. -func (c *client) forget(id int64) { - c.mu.Lock() - delete(c.pending, id) - c.mu.Unlock() + return c.t.call(method, params) } // notify sends a JSON-RPC message with no reply expected. func (c *client) notify(method string, params any) error { - note := rpcNotification{JSONRPC: "2.0", Method: method, Params: params} - data, err := json.Marshal(note) - if err != nil { - return fmt.Errorf("mcp: encode %s: %w", method, err) - } - data = append(data, '\n') - _, err = c.stdin.Write(data) - return err + return c.t.notify(method, params) } -// Close stops the subprocess, giving it a moment to exit on its own before it -// is killed outright. +// Close stops the subprocess, or ends the HTTP session, depending on which +// transport this client was opened with. func (c *client) Close() error { - c.stdin.Close() - - done := make(chan error, 1) - go func() { done <- c.cmd.Wait() }() - - select { - case <-done: - case <-time.After(2 * time.Second): - if c.cmd.Process != nil { - c.cmd.Process.Kill() - } - <-done - } - return nil + return c.t.Close() } // initialize performs the handshake every MCP server requires before any diff --git a/internal/mcp/stdio_transport.go b/internal/mcp/stdio_transport.go new file mode 100644 index 0000000..820d6f8 --- /dev/null +++ b/internal/mcp/stdio_transport.go @@ -0,0 +1,195 @@ +package mcp + +import ( + "bufio" + "encoding/json" + "fmt" + "io" + "os" + "os/exec" + "strings" + "sync" + "time" +) + +// stdioTransport is one running local MCP server subprocess, spoken to over +// newline-delimited JSON-RPC on its stdin and stdout. +// +// A single background goroutine owns the read side and demultiplexes +// responses to the call that is waiting for each one by ID, which is what +// lets a call time out without corrupting a read another call still needs: +// the reader never stops, it only stops being waited on. +type stdioTransport struct { + cmd *exec.Cmd + stdin io.WriteCloser + + mu sync.Mutex + nextID int64 + pending map[int64]chan rpcResponse + readErr error +} + +// newStdioTransport launches a server's subprocess and begins reading its +// output. +func newStdioTransport(s Server) (*stdioTransport, error) { + cmd := exec.Command(s.Command, s.Args...) + cmd.Env = append([]string{"PATH=" + os.Getenv("PATH")}, s.Env...) + + stdin, err := cmd.StdinPipe() + if err != nil { + return nil, fmt.Errorf("mcp: open stdin: %w", err) + } + stdout, err := cmd.StdoutPipe() + if err != nil { + return nil, fmt.Errorf("mcp: open stdout: %w", err) + } + // Stderr is discarded rather than captured: a server's diagnostic chatter + // is not a resource and is not shown to the model, and capturing it + // without a bound would be one more place a hung or chatty server could + // grow without limit. + cmd.Stderr = io.Discard + + if err := cmd.Start(); err != nil { + return nil, fmt.Errorf("mcp: start %s: %w", s.Command, err) + } + + t := &stdioTransport{cmd: cmd, stdin: stdin, pending: make(map[int64]chan rpcResponse)} + go t.readLoop(stdout) + return t, nil +} + +// readLoop reads one JSON-RPC message per line for the life of the +// subprocess, and delivers each to whichever call is waiting for its ID. A +// line this client cannot parse, or one carrying no numeric ID - a +// notification a server sent unasked - is skipped rather than treated as a +// fault: this client has nothing it is waiting to do with either. +func (t *stdioTransport) readLoop(r io.Reader) { + scanner := bufio.NewScanner(r) + scanner.Buffer(make([]byte, 0, 64*1024), 16<<20) + + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" { + continue + } + var resp rpcResponse + if json.Unmarshal([]byte(line), &resp) != nil { + continue + } + var id int64 + if json.Unmarshal(resp.ID, &id) != nil { + continue + } + + t.mu.Lock() + ch, ok := t.pending[id] + if ok { + delete(t.pending, id) + } + t.mu.Unlock() + + if ok { + ch <- resp + } + } + + err := fmt.Errorf("mcp: the server's output ended") + if scanner.Err() != nil { + err = fmt.Errorf("mcp: read: %w", scanner.Err()) + } + + t.mu.Lock() + t.readErr = err + pending := t.pending + t.pending = make(map[int64]chan rpcResponse) + t.mu.Unlock() + + for _, ch := range pending { + close(ch) + } +} + +// call sends a request and waits for its matching response, or for +// callTimeout to pass. +// +// A call that times out drops its own slot so the (possibly still arriving) +// response is discarded by readLoop rather than delivered to no one, but it +// does not stop the subprocess: a slow answer is not evidence the server is +// broken, and the next call gets its own fresh wait. +func (t *stdioTransport) call(method string, params any) (json.RawMessage, error) { + t.mu.Lock() + t.nextID++ + id := t.nextID + ch := make(chan rpcResponse, 1) + t.pending[id] = ch + t.mu.Unlock() + + req := rpcRequest{JSONRPC: "2.0", ID: id, Method: method, Params: params} + data, err := json.Marshal(req) + if err != nil { + t.forget(id) + return nil, fmt.Errorf("mcp: encode %s: %w", method, err) + } + data = append(data, '\n') + + if _, err := t.stdin.Write(data); err != nil { + t.forget(id) + return nil, fmt.Errorf("mcp: write %s: %w", method, err) + } + + select { + case resp, ok := <-ch: + if !ok { + t.mu.Lock() + readErr := t.readErr + t.mu.Unlock() + return nil, fmt.Errorf("mcp: %s: %w", method, readErr) + } + if resp.Error != nil { + return nil, fmt.Errorf("mcp: %s: %s (code %d)", method, resp.Error.Message, resp.Error.Code) + } + return resp.Result, nil + + case <-time.After(callTimeout): + t.forget(id) + return nil, fmt.Errorf("mcp: %s timed out after %s", method, callTimeout) + } +} + +// forget removes a call's slot without waiting for it further. +func (t *stdioTransport) forget(id int64) { + t.mu.Lock() + delete(t.pending, id) + t.mu.Unlock() +} + +// notify sends a JSON-RPC message with no reply expected. +func (t *stdioTransport) notify(method string, params any) error { + note := rpcNotification{JSONRPC: "2.0", Method: method, Params: params} + data, err := json.Marshal(note) + if err != nil { + return fmt.Errorf("mcp: encode %s: %w", method, err) + } + data = append(data, '\n') + _, err = t.stdin.Write(data) + return err +} + +// Close stops the subprocess, giving it a moment to exit on its own before it +// is killed outright. +func (t *stdioTransport) Close() error { + t.stdin.Close() + + done := make(chan error, 1) + go func() { done <- t.cmd.Wait() }() + + select { + case <-done: + case <-time.After(2 * time.Second): + if t.cmd.Process != nil { + t.cmd.Process.Kill() + } + <-done + } + return nil +} diff --git a/orcli.json.example b/orcli.json.example index 7a8f1d5..45cc808 100644 --- a/orcli.json.example +++ b/orcli.json.example @@ -44,6 +44,13 @@ "env": { "EXAMPLE_API_KEY": "your-example-key" } + }, + { + "name": "remote-example", + "url": "https://mcp.example.com/mcp", + "headers": { + "Authorization": "Bearer your-remote-server-token" + } } ] },