mcp

type AuthTokenFunc ¶

type AuthTokenFunc func(ctx context.Context) (string, error)

AuthTokenFunc returns a Bearer token to attach to outbound MCP requests. Returning ("", nil) means "no auth header" — typical for in-cluster trust networks. The function is invoked PER REQUEST so the OAuthFlow can transparently refresh expired tokens.

type CallToolParams ¶

type CallToolParams struct {
	Name      string          `json:"name"`
	Arguments json.RawMessage `json:"arguments,omitempty"`
}

CallToolParams is the params field of a tools/call request.

type CallToolResult ¶

type CallToolResult struct {
	Content []ToolContent `json:"content"`
	IsError bool          `json:"isError,omitempty"`
}

CallToolResult is the result field of a tools/call response.

type Client ¶

type Client interface {
	Initialize(ctx context.Context, info ClientInfo) (*InitializeResult, error)
	Initialized(ctx context.Context) error
	ListTools(ctx context.Context) ([]MCPToolDescriptor, error)
	CallTool(ctx context.Context, name string, args json.RawMessage) (*CallToolResult, error)
	Close() error
}

Client speaks the four MCP RPCs Phase 1 needs. It wraps a Transport with request/response demultiplexing: each call gets a fresh monotonic ID, sends, then waits for the matching response.

Concurrency: CallTool may be invoked from multiple goroutines; the internal demultiplexer correctly routes responses by ID. Initialize must be the first call on a fresh Client — the Server lifecycle in server.go enforces that.

type ClientInfo ¶

type ClientInfo struct {
	Name    string `json:"name"`
	Version string `json:"version"`
}

ClientInfo is the payload sent in the initialize request.

type HTTPTransport ¶

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

HTTPTransport speaks the MCP Streamable HTTP transport: every frame is a JSON-RPC 2.0 POST to a single endpoint. The server MAY upgrade the response to text/event-stream when it wants to emit multiple frames (e.g., progress events from a long tool call).

Caller is responsible for providing the *http.Client — typically security.EgressClientFromContext(ctx) so requests ride the Forge egress allowlist. HTTPTransport never constructs its own client.

Concurrency: Send is safe to call from multiple goroutines. Recv is intended to be called from a single demultiplexer goroutine in the Client (a future refactor could lift this restriction if server-initiated frames become common).

Backpressure (review B3): the response queue is bounded. When it fills, push() blocks (it does NOT silently drop the oldest frame — the previous behavior, which orphaned the in-flight CallTool whose response was discarded). If the queue stays full longer than overflowTimeout (default 30s) the transport is closed with ErrTransportUnavailable so callers fail loudly instead of hanging on their per-ID response channels.

func NewHTTPTransport(url string, httpClient *http.Client, authFn AuthTokenFunc) (*HTTPTransport, error)

func (h *HTTPTransport) Close() error

func (h *HTTPTransport) Recv(ctx context.Context) (JSONRPCMessage, error)

func (h *HTTPTransport) Send(ctx context.Context, msg JSONRPCMessage) error

func (h *HTTPTransport) SessionID() string

type InitializeParams ¶

type InitializeParams struct {
	ProtocolVersion string         `json:"protocolVersion"`
	Capabilities    map[string]any `json:"capabilities,omitempty"`
	ClientInfo      ClientInfo     `json:"clientInfo"`
}

InitializeParams is the params field of an initialize request.

type InitializeResult ¶

type InitializeResult struct {
	ProtocolVersion string         `json:"protocolVersion"`
	Capabilities    map[string]any `json:"capabilities,omitempty"`
	ServerInfo      ServerInfo     `json:"serverInfo"`
}

InitializeResult is the result field of an initialize response.

type JSONRPCError ¶

type JSONRPCError struct {
	Code    int             `json:"code"`
	Message string          `json:"message"`
	Data    json.RawMessage `json:"data,omitempty"`
}

JSONRPCError is the error object on a JSON-RPC 2.0 response.

type JSONRPCMessage ¶

type JSONRPCMessage struct {
	Jsonrpc string          `json:"jsonrpc"`
	ID      *json.Number    `json:"id,omitempty"`
	Method  string          `json:"method,omitempty"`
	Params  json.RawMessage `json:"params,omitempty"`
	Result  json.RawMessage `json:"result,omitempty"`
	Error   *JSONRPCError   `json:"error,omitempty"`
}

JSONRPCMessage is the union frame for JSON-RPC 2.0 over MCP.

A single frame can be:

• request: ID set, Method set, Params optional
• response: ID set, Result OR Error set (never both)
• notification: Method set, ID nil

We intentionally keep Result/Params as json.RawMessage so we don't lose fidelity round-tripping schemas to the LLM function-calling layer or losing precision on numeric IDs.

func (m JSONRPCMessage) Validate() error

type ListToolsResult ¶

type ListToolsResult struct {
	Tools []MCPToolDescriptor `json:"tools"`
}

ListToolsResult is the result field of a tools/list response.

type MCPToolDescriptor ¶

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

MCPToolDescriptor is the parsed form of a single entry in the tools/list response. The Phase 1 adapter (Commit 4) wraps each descriptor in a tools.Tool that the LLM executor consumes directly.

InputSchema is a raw JSON Schema (draft-07) used both for runtime validation and as the parameter spec advertised to the LLM. We keep it as json.RawMessage rather than a parsed Go type so we never lose fidelity round-tripping it to the OpenAI/Anthropic function-calling shape.

func FilterTools(descs []MCPToolDescriptor, f types.MCPToolFilter) []MCPToolDescriptor

type Manager ¶

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

Manager owns the set of Server instances declared in a forge.yaml mcp.servers[] block. Manager.Start spawns one goroutine per server (parallel startup); Start returns only after every server has reached Ready or terminally Failed.

If any Required=true server fails, Start cancels the shared child context — all other servers tear down — and returns a non-nil error. The caller (runner.go) propagates this upward, exiting the agent with non-zero status (K8s sees CrashLoopBackOff).

Tools() aggregates discovered tools across Ready servers into a flat list suitable for registry registration. The caller is responsible for wrapping each descriptor in an adapter.MCPTool — done in runner.go to avoid a circular dependency between forge-core/mcp and forge-core/tools/adapters.

func NewManager(cfg types.MCPConfig, deps ManagerDeps) (*Manager, error)

func (m *Manager) Servers() map[string]*Server

func (m *Manager) Start(ctx context.Context) error

func (m *Manager) Stop() error

func (m *Manager) Tools() []ToolHandle

type ManagerDeps ¶

type ManagerDeps struct {
	// HTTPClient is the egress-controlled client used for all MCP
	// HTTP traffic. Required — runner.go passes
	// security.EgressClientFromContext(ctx) here.
	HTTPClient *http.Client

	// Logger is used for non-audit warnings. nil → no-op.
	Logger ServerLogger

	// Audit emits mcp_server_* / mcp_tool_* events. Required for
	// production wiring; tests may pass nil.
	Audit *runtime.AuditLogger

	// OAuth is the shared OAuthFlow used by any server with
	// auth.type=oauth. Required when at least one such server exists
	// in cfg; otherwise may be nil.
	OAuth *OAuthFlow
}

ManagerDeps groups the dependencies Manager hands to each Server.

type OAuthFlow ¶

type OAuthFlow struct {
	// RefreshWindow is the slack before expiry at which BearerToken
	// proactively refreshes. Default 60s.
	RefreshWindow time.Duration

	// RefreshTimeout caps each /token call. Default 30s.
	// Decoupled from any caller's context — see the type docstring.
	RefreshTimeout time.Duration

	// HTTPClient is used for token-endpoint requests. nil → a
	// defaulting *http.Client with no Transport-level timeout (the
	// per-call ctx supplies the bound). Production wiring should pass
	// the egress-controlled client (security.EgressClient) so token
	// endpoints ride the same allowlist as MCP traffic.
	HTTPClient *http.Client

	// AuditFn is called on every refresh attempt (success and
	// failure). nil means no audit — typical for Login at laptop time.
	AuditFn func(server string, ok bool, reason string)

	// BrowserOpener opens a URL in the operator's browser during
	// Login. REQUIRED — Login returns an error if nil. We
	// deliberately do NOT provide a default that shells out to
	// xdg-open/open/start because forge-core/mcp must remain free
	// of an `os/exec` import (spec §4.6, review B4). The browser-
	// opening glue lives in forge-cli/cmd/mcp_login.go where
	// `os/exec` is permitted because that code never ships in the
	// runtime call graph reachable from `forge run`. Tests inject
	// a no-op or a redirect-driving helper.
	BrowserOpener func(url string) error
	// contains filtered or unexported fields
}

OAuthFlow implements OAuth 2.1 with PKCE for MCP servers, sharing the encrypted token store with the existing llm/oauth package (decision §3.6 of the recommendations doc). MCP tokens live under a separate key namespace so they cannot collide with LLM provider tokens.

Two flows:

Login(ctx, name, cfg) — laptop-time, interactive. Generates PKCE,
opens a loopback listener, opens the operator's browser at the
authorization endpoint, exchanges the returned code for tokens,
persists them in the encrypted store.

BearerToken(ctx, name, cfg) — runtime, automatic. Loads tokens,
refreshes if within RefreshWindow of expiry, returns the
access_token. Refresh failure (invalid_grant / expired_token)
surfaces as ErrTokenRevoked.

The refresh path is concurrency-safe via per-name singleflight — 100 concurrent BearerToken calls produce one /token call.

IMPORTANT (review B2): the singleflight goroutine uses its OWN background-derived context with a hard RefreshTimeout cap — it is NOT bound to the leader caller's ctx. Otherwise a misbehaving IdP hangs the goroutine indefinitely, leaks the inFly slot, and wedges every subsequent caller for the same server. Caller-ctx-cancel unblocks the caller's wait on <-grp.done but never affects the in-flight refresh.

func NewOAuthFlow() *OAuthFlow

func (f *OAuthFlow) BearerToken(ctx context.Context, name string, cfg OAuthServerConfig) (string, error)

func (f *OAuthFlow) Login(ctx context.Context, name string, cfg OAuthServerConfig) error

func (f *OAuthFlow) Logout(name string) error

type OAuthServerConfig ¶

type OAuthServerConfig struct {
	ClientID     string
	Scopes       []string
	AuthorizeURL string
	TokenURL     string
}

OAuthServerConfig captures the per-server OAuth knobs needed at flow time. Plays the role of types.MCPAuth without the YAML tags, to keep this package importable from cmd/ without a dependency on the types package shape changing.

type Server ¶

type Server struct {
	Name string
	Spec types.MCPServer
	// contains filtered or unexported fields
}

Server wraps a single MCP server in its lifecycle state machine. One Server per entry in forge.yaml mcp.servers[]. Owned by the Manager (Commit 4); not constructed directly by application code.

Lifecycle is driven by Run(ctx): a single goroutine progresses through Connecting → Initializing → Discovering → Ready, then sits in Ready until ctx is cancelled or a transport failure pushes it into Degraded → Reconnecting → (Failed or Initializing again).

Tools is safe to call only AFTER state has reached Ready at least once.

func NewServer(spec types.MCPServer, deps ServerDeps) (*Server, error)

func (s *Server) Client() Client

func (s *Server) Failed() <-chan struct{}

func (s *Server) Ready() <-chan struct{}

func (s *Server) Run(ctx context.Context) error

func (s *Server) State() ServerState

func (s *Server) Tools() []MCPToolDescriptor

type ServerDeps ¶

type ServerDeps struct {
	HTTPClient *http.Client // injected; never default
	Logger     ServerLogger // nil → no-op
	Audit      *runtime.AuditLogger
	OAuth      *OAuthFlow // nil → no OAuth servers in this config
}

ServerDeps bundles the dependencies a Server needs. Pulled out so the Manager can hand the same set to every Server without verbose constructor signatures.

type ServerInfo ¶

type ServerInfo struct {
	Name    string `json:"name"`
	Version string `json:"version"`
}

ServerInfo is the payload received in the initialize response.

type ServerLogger ¶

type ServerLogger interface {
	Info(msg string, fields map[string]any)
	Warn(msg string, fields map[string]any)
	Error(msg string, fields map[string]any)
}

ServerLogger is a minimal logger interface — keeps the package free of a hard dependency on a particular logger implementation. The runtime.Logger interface satisfies it structurally.

type ServerState ¶

type ServerState int

ServerState tracks a single MCP server's position in its lifecycle state machine. See server.go (Commit 3) for the driver and server_state.go for the legal transition table.

The String() form is what surfaces in `forge mcp list` output and in audit events; keep it stable — operators grep on it.

const (
	// StateConfigured — parsed from forge.yaml, not yet started. Initial
	// state set by NewServer.
	StateConfigured ServerState = iota

	// StateConnecting — opening the underlying Transport (HTTP dial,
	// OAuth refresh). Next: Initializing | Failed.
	StateConnecting

	// StateInitializing — performing the JSON-RPC initialize handshake
	// (including the "initialized" notification). Hard-fails on
	// protocol-version mismatch. Next: Discovering | Failed.
	StateInitializing

	// StateDiscovering — calling tools/list and validating each
	// descriptor's input schema as JSON Schema draft-07. A malformed
	// schema fails the SERVER, not the LLM call. Next: Ready | Failed.
	StateDiscovering

	// StateReady — handshake complete, tools registered, server idle.
	// Next: Calling | Stopped.
	StateReady

	// StateCalling — at least one tools/call is in flight. Concurrent
	// calls do not reset the state; we transition back to Ready only
	// when all in-flight calls resolve. Next: Ready | Degraded |
	// Stopped.
	StateCalling

	// StateDegraded — transient transport error mid-call. Will attempt
	// to reconnect per the backoff schedule. Next: Reconnecting |
	// Stopped.
	StateDegraded

	// StateReconnecting — re-running the connect+initialize+discover
	// chain after backoff. Next: Initializing (on transport open) |
	// Failed (on backoff exhaustion).
	StateReconnecting

	// StateFailed — terminal failure. Required=true servers cause the
	// Manager to cancel its parent context (i.e., the agent exits).
	// Required=false servers simply have their tools removed from the
	// registry. Next: Stopped.
	StateFailed

	// StateStopped — terminal. Reached after ctx cancel or after Failed.
	// No outbound transitions.
	StateStopped
)

func (s ServerState) String() string

type ToolContent ¶

type ToolContent struct {
	Type     string          `json:"type"` // "text" | "image" | "resource"
	Text     string          `json:"text,omitempty"`
	MimeType string          `json:"mimeType,omitempty"`
	Data     string          `json:"data,omitempty"`     // base64 for image/resource
	Resource json.RawMessage `json:"resource,omitempty"` // resource reference
}

ToolContent is one piece of a tool's response payload.

type ToolHandle ¶

type ToolHandle struct {
	Server     string
	Descriptor MCPToolDescriptor
	Client     Client
}

Tools returns a flat list of (server, descriptor, client) tuples for every Ready server. Callers wrap each tuple in an adapters.MCPTool. Order across servers is deterministic by server name; order within a server is the discovery order.

type Transport ¶

type Transport interface {
	// Send writes one frame to the wire. For HTTP this is a single
	// POST; for stdio it would be a single line of JSON. Returns when
	// the frame has been handed off; does NOT block waiting for a
	// response.
	Send(ctx context.Context, msg JSONRPCMessage) error

	// Recv blocks until the next frame arrives or ctx is cancelled.
	// Returns ErrClosed once Close has been called and the queue has
	// drained.
	Recv(ctx context.Context) (JSONRPCMessage, error)

	// Close releases all resources. Idempotent. Frames in flight at
	// Close time are dropped; callers blocked on Recv are unblocked
	// with ErrClosed.
	Close() error
}

Transport carries JSON-RPC frames between Forge and a single MCP server. Implementations are responsible for the wire encoding, per-call timeouts, and propagating ctx cancellation.

Phase 1 has exactly one implementation: HTTPTransport in transport_http.go. A future stdio implementation would live in a separate file behind a feature gate (per the deferred-stdio decision); the interface itself does not change.

Concurrency: Send and Recv MAY be called concurrently from different goroutines. Send MUST be safe to call concurrently with itself. Recv is typically driven by a single goroutine (the Client's response demultiplexer).