mcp

package
v0.2.6 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 15, 2026 License: Apache-2.0 Imports: 17 Imported by: 0

Documentation

Overview

Package mcp implements a Model Context Protocol (MCP) client for the rysh CLI.

It lets rysh agents/humanoids consume tools exposed by external MCP servers over two transports:

  • stdio: a child process speaking newline-delimited JSON-RPC on stdin/stdout.
  • Streamable HTTP: a long-running HTTP service whose /mcp endpoint accepts JSON-RPC POSTs and replies with application/json or text/event-stream.

Discovered tools are adapted to the shared tools.ToolExecutor interface (see executor.go) and registered into the agent tool registry by the Manager (see manager.go), so they flow through the existing Anthropic tool-use bridge, approval gates, and audit path unchanged.

The wire types mirror the in-repo reference servers under rysh-mcp-samples (mcp-server = stdio, mcp-rest-api-wrapper = Streamable HTTP).

Index

Constants

View Source 
const (
	TransportStdio = "stdio"
	TransportHTTP  = "http"
)

TransportStdio / TransportHTTP are the supported transport identifiers.

View Source 
const ProtocolVersion = "2024-11-05"

ProtocolVersion is the MCP revision this client advertises during the initialize handshake. It matches the version spoken by the reference servers in rysh-mcp-samples.

Variables

View Source 
var HeartbeatInterval = 30 * time.Second

HeartbeatInterval is the cadence at which the manager probes each connected server with a lightweight tools/list call. Follow-up 6b. Exposed as a var so tests can shrink it.

View Source 
var HeartbeatTimeout = 5 * time.Second

HeartbeatTimeout bounds a single probe RPC. Follow-up 6b.

View Source 
var MaxRestartAttemptsPerSession = 20

MaxRestartAttemptsPerSession bounds how many times a single server will be auto-reconnected after a transport failure. Beyond this, the operator must explicitly trigger `##mcp restart <name>`. Follow-up item 6.

Functions

func SaveStore 

func SaveStore(workDir string, defs []ServerDef) error

SaveStore writes server definitions, creating the .rysh directory if needed. The write is atomic (temp file + rename) so a crash never leaves a partial store.

func StorePath 

func StorePath(workDir string) string

StorePath returns the location of the MCP server store for a project rooted at workDir: <workDir>/.rysh/mcp.json.

Types

type Client 

type Client struct {
	// contains filtered or unexported fields
}

Client is a protocol-level MCP client over a single transport. It performs the initialize handshake, lists tools (with pagination), and calls tools. It is safe for concurrent use.

func (*Client) CallTool 

func (c *Client) CallTool(ctx context.Context, name string, args json.RawMessage) (*callToolResult, error)

CallTool invokes a tool by its server-side name with raw JSON arguments.

func (*Client) Close 

func (c *Client) Close() error

Close shuts down the underlying transport.

func (*Client) Initialize 

func (c *Client) Initialize(ctx context.Context) error

Initialize performs the MCP handshake: it sends "initialize" and, on success, the "notifications/initialized" acknowledgement.

func (*Client) ListTools 

func (c *Client) ListTools(ctx context.Context) ([]Tool, error)

ListTools returns every tool the server advertises, following nextCursor pagination so large catalogs are fully enumerated.

func (*Client) ServerInfo 

func (c *Client) ServerInfo() (name, version string)

ServerInfo returns the name/version reported by the server (valid after Initialize).

func (*Client) SetToolsChangedHandler 

func (c *Client) SetToolsChangedHandler(fn func())

SetToolsChangedHandler registers a callback invoked when the server announces notifications/tools/list_changed (stdio transport only).

type Manager 

type Manager struct {
	// contains filtered or unexported fields
}

Manager owns the set of connected MCP servers for a session and registers their tools into a shared tools.ToolRegistry. It is safe for concurrent use.

Registration target matters: the registry handed in is the *shared* agent registry that every pane/agent clones at creation. Connecting at startup (Bootstrap) therefore makes MCP tools available to all panes; a live `##mcp add` reaches panes/agents created afterward.

func NewManager 

func NewManager(registry *sharedtools.ToolRegistry, workDir string) *Manager

NewManager creates a Manager that registers tools into the given shared registry. workDir locates the persisted store and is the working directory for stdio child processes.

func (*Manager) AddServer 

func (m *Manager) AddServer(ctx context.Context, def ServerDef) (int, error)

AddServer validates, persists, connects, and registers a single server. The definition is persisted even when the connection fails, so a transiently down server is retried on the next startup. Returns the number of tools registered.

func (*Manager) AddServerScoped 

func (m *Manager) AddServerScoped(ctx context.Context, def ServerDef, target ScopeTarget) (int, error)

AddServerScoped adds a server whose tools register into target's registry. The scope key is persisted on the def (re-applied at global on restart, as forge); the runtime target is recorded so reconnects re-register into the same scope.

func (*Manager) Bootstrap 

func (m *Manager) Bootstrap(ctx context.Context)

Bootstrap loads persisted server definitions and connects to all of them concurrently. It is intended to run once at startup, before any pane clones the shared registry. Errors connecting individual servers are logged, not fatal.

func (*Manager) Close 

func (m *Manager) Close()

func (*Manager) ConnectAll 

func (m *Manager) ConnectAll(ctx context.Context, defs []ServerDef)

ConnectAll connects to every definition concurrently and registers tools. It records per-server status (including failures) and never returns an error; inspect List() for results.

func (*Manager) Count 

func (m *Manager) Count() int

Count returns the number of known servers.

func (*Manager) GlobalTarget 

func (m *Manager) GlobalTarget() ScopeTarget

GlobalTarget is the default scope target: the shared session-wide registry.

func (*Manager) List 

func (m *Manager) List() []ServerStatus

List returns a sorted snapshot of all known servers.

func (*Manager) MarkUnhealthy 

func (m *Manager) MarkUnhealthy(name string, cause error) bool

MarkUnhealthy records that a server's transport failed and triggers an async reconnect. Safe to call concurrently; only ONE reconnect goroutine ever runs per server at a time. Returns true when a fresh reconnect was scheduled, false when one was already in flight or the server has exhausted its session cap.

func (*Manager) Reconnect 

func (m *Manager) Reconnect(ctx context.Context, name string) (int, error)

Reconnect re-establishes a configured server using its stored definition.

func (*Manager) RemoveServer 

func (m *Manager) RemoveServer(name string) error

RemoveServer disconnects a server, unregisters its tools, and drops it from the persisted store.

func (*Manager) ReplayScope 

func (m *Manager) ReplayScope(ctx context.Context, scopeKey string, target ScopeTarget)

ReplayScope re-establishes every persisted MCP server that was added at the given scope key, registering its tools into target's registry. A Tab/Lane/PaneGroup/Pane actor calls it when it restores with a stable scope-instance id (mirrors forge.Manager.ReplayScope), so a server added with `--scope lane` lands back on the same lane after a restart. No-op for the global scope (Bootstrap handles that). Connects synchronously, so callers that run on an actor mailbox should invoke it in a goroutine.

func (*Manager) SetStatusEmitter 

func (m *Manager) SetStatusEmitter(fn func(StatusEvent))

SetStatusEmitter installs a callback fired on every server state transition. It must be cheap and non-blocking — it runs on the manager's own goroutines (the connect waves, the heartbeat, the reconnect loop). Call it once before Bootstrap so the initial connect wave reports too. Follow-up 6b.

func (*Manager) StartHeartbeat 

func (m *Manager) StartHeartbeat()

StartHeartbeat launches the periodic liveness probe (follow-up 6b). Called from Bootstrap after the initial connect wave. Safe to call multiple times: the second call is a no-op while the first is still running.

func (*Manager) StopHeartbeat 

func (m *Manager) StopHeartbeat()

StopHeartbeat signals the heartbeat goroutine to exit and waits for it to finish. Idempotent.

func (*Manager) ToolsOf 

func (m *Manager) ToolsOf(name string) ([]ToolInfo, bool)

ToolsOf returns the registered tools for a server.

func (*Manager) UnregisterScope 

func (m *Manager) UnregisterScope(scopeKey string)

Close disconnects every server and unregisters all MCP tools. UnregisterScope disconnects and unregisters every MCP server whose tools were registered at the given scope key, when that scope instance is torn down. Servers added without a scope are global and are never auto-removed here.

func (*Manager) WorkDir 

func (m *Manager) WorkDir() string

WorkDir returns the project root used for the store and stdio child processes.

type ScopeTarget 

type ScopeTarget struct {
	Key      string
	Registry *sharedtools.ToolRegistry
}

ScopeTarget tells AddServerScoped which registry to register a server's tools into and a stable key for that scope instance. Mirrors forge.ScopeTarget; the caller (which knows the scope hierarchy) resolves it.

type ServerDef 

type ServerDef struct {
	Name      string `json:"name"`
	Transport string `json:"transport"` // "stdio" | "http"

	// stdio transport
	Command string   `json:"command,omitempty"`
	Args    []string `json:"args,omitempty"`
	Env     []string `json:"env,omitempty"` // "KEY=VALUE" entries added to the child env

	// http (Streamable HTTP) transport
	URL     string            `json:"url,omitempty"`
	Headers map[string]string `json:"headers,omitempty"` // e.g. {"Authorization": "Bearer …"}

	// Cross-cutting policy
	Prefix           *string `json:"prefix,omitempty"`            // tool-name prefix; nil ⇒ "<name>_"; "" ⇒ none
	RequiresApproval bool    `json:"requires_approval,omitempty"` // gate every call from this server
	MaxTools         int     `json:"max_tools,omitempty"`         // cap registered tools (0 ⇒ defaultMaxTools)
	Scope            string  `json:"scope,omitempty"`             // scope key the server's tools register at ("" ⇒ global)
}

ServerDef is a persisted MCP server definition. It is stored as JSON in the project-local .rysh/mcp.json file (alongside .rysh/agents/), so MCP servers are a per-project asset that can be checked into git.

func LoadStore 

func LoadStore(workDir string) ([]ServerDef, error)

LoadStore reads persisted server definitions. A missing file yields an empty slice with no error (a project simply has no MCP servers configured yet).

func ParseAddArgs 

func ParseAddArgs(args []string) (ServerDef, error)

ParseAddArgs parses the `##mcp add` argument list (everything after "add") into a ServerDef.

Grammar: 

<name> http  <url>       [flags...]
<name> stdio <cmd> [args... up to the first --flag] [flags...]

Flags: --approve, --prefix <p>, --max-tools <n>, --header <k:v> (http, repeatable), --env <k=v> (stdio, repeatable). For stdio, command arguments beginning with "--" cannot be expressed (they read as flags); single-dash args like "-y" are fine.

func (ServerDef) Detail 

func (d ServerDef) Detail() string

Detail renders a one-line locator for a server definition (url or command).

func (ServerDef) Validate 

func (d ServerDef) Validate() error

Validate checks a definition is internally consistent before connecting.

type ServerPhase 

type ServerPhase string

ServerPhase enumerates the externally-visible lifecycle states surfaced to observers through the StatusEmitter. Follow-up 6b. 

const (
	PhaseConnected    ServerPhase = "connected"
	PhaseReconnecting ServerPhase = "reconnecting"
	PhaseGivenUp      ServerPhase = "given_up"
	PhaseDisconnected ServerPhase = "disconnected"
	PhaseRemoved      ServerPhase = "removed"
)

type ServerStatus 

type ServerStatus struct {
	Name       string
	Transport  string
	Connected  bool
	Registered int
	Discovered int
	Error      string
	Detail     string // url (http) or "command args" (stdio)
}

ServerStatus is a snapshot of one server for `##mcp list`.

type StatusEvent 

type StatusEvent struct {
	Server  string
	Phase   ServerPhase
	Attempt int    // current reconnect attempt (1-based) when Phase==reconnecting / given_up
	Max     int    // MaxRestartAttemptsPerSession at emit time
	Detail  string // human-readable detail / last error (may be empty)
}

StatusEvent is emitted on every MCP server state transition when a StatusEmitter is installed. It is transport-agnostic on purpose — the host wires it to whatever surface it likes; the CLI publishes it on the session-global mcp.status NATS subject for the TUI footer. Follow-up 6b.

type Tool 

type Tool struct {
	Name        string          `json:"name"`
	Description string          `json:"description"`
	InputSchema json.RawMessage `json:"inputSchema"`
}

Tool is a tool advertised by an MCP server. InputSchema is kept as raw JSON so it can be handed to tools.ToolSpec.Parameters (a JSON Schema) without a lossy round-trip through a typed struct.

type ToolInfo 

type ToolInfo struct {
	RemoteName     string
	RegisteredName string
	Description    string
}

ToolInfo describes one registered MCP tool for `##mcp tools <name>`.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.