From d8ea627bda591ad77c09e976b2ab51ad82332943 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 19:58:49 +0000 Subject: [PATCH] MCP: support remote servers over Streamable HTTP (#62) internal/mcp previously only reached a configured server by launching it as a local subprocess over stdio. This adds the current recommended remote transport, Streamable HTTP (MCP spec 2025-03-26), while keeping local stdio support unchanged. Split the transport out behind a small interface (call/notify/Close) so client's own protocol methods (initialize, listResources, readResource) are written once and work unchanged against either kind of server - the existing stdio transport moved into its own file (stdio_transport.go) with no behavior change (its own tests pass unmodified), and a new httpTransport (http_transport.go) implements the remote side: - 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 - a server may legally answer either way, and both are handled. - Tracks the Mcp-Session-Id a server assigns on initialize and sends it back on every later request in the same session. - Best-effort session termination (DELETE) on Close, matching the spec's SHOULD without making a remote server's non-cooperation a local failure. - Deliberately does not open a standalone server-initiated SSE stream (a bare GET to the same URL) - this client does not act on an unsolicited server notification over stdio either, so there is nothing this narrows that today's stdio behavior does not already narrow the same way. Config (internal/config/mcp.go): a server now names `command` (local) or `url` + optional `headers` (remote), never both and never neither - refused by name at configuration read time, the same way every other misconfiguration in this block already is. cmd/orcli/ask.go's newMCPManager passes URL/Headers through to mcp.Server, the one place this package's own dependency on internal/mcp is named. orcli.json.example and README.md document the remote shape side by side with the existing local one. `make test` passes clean. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Nxr4mW9rMNzn9kThjdJpZt --- README.md | 23 ++- cmd/orcli/ask.go | 5 +- cmd/orcli/ask_test.go | 24 +++ internal/config/load_test.go | 7 +- internal/config/mcp.go | 68 ++++--- internal/config/mcp_test.go | 59 ++++++ internal/mcp/http_transport.go | 252 +++++++++++++++++++++++++ internal/mcp/http_transport_test.go | 272 +++++++++++++++++++++++++++ internal/mcp/mcp.go | 282 +++++++++------------------- internal/mcp/stdio_transport.go | 195 +++++++++++++++++++ orcli.json.example | 7 + 11 files changed, 976 insertions(+), 218 deletions(-) create mode 100644 internal/mcp/http_transport.go create mode 100644 internal/mcp/http_transport_test.go create mode 100644 internal/mcp/stdio_transport.go 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" + } } ] },