registry

type ClientSession ¶

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

ClientSession is one client connection's set of downstream sessions and the domain.Dispatcher for that connection. It is safe for concurrent Dispatch. Close must not run concurrently with Dispatch: the composition root drains in-flight calls before closing a client session (design §5 shutdown order).

func (cs *ClientSession) Close() error

func (cs *ClientSession) Dispatch(ctx context.Context, call *domain.Call) (*domain.Result, error)

type DownstreamSession ¶

type DownstreamSession interface {
	Session
	// CallTool invokes a downstream tool by its un-namespaced name with the given
	// raw JSON arguments, returning the result for the outbound pipeline.
	CallTool(ctx context.Context, tool string, args json.RawMessage) (*domain.Result, error)
	// ListTools returns one page of the downstream's tools plus the next
	// pagination cursor ("" when the listing is complete).
	ListTools(ctx context.Context, cursor string) (tools []ToolInfo, next string, err error)
	// Ping checks that the downstream is responsive.
	Ping(ctx context.Context) error
}

DownstreamSession is a live session to one downstream MCP server. It extends the pool's Session (which only needs Close) with the operations the dispatcher performs against a downstream: calling tools, discovering them, and pinging for liveness. It is transport-agnostic — both the stdio (subprocess) and remote-HTTP session types implement it.

type Factory ¶

type Factory interface {
	New(ctx context.Context) (Session, error)
}

Factory creates a new downstream session, spawning whatever transport backs it (an stdio subprocess, a remote HTTP connection).

type HTTPConfig ¶

type HTTPConfig struct {
	// URL is the downstream's MCP endpoint (the Streamable HTTP endpoint).
	URL string
	// Headers are connection-level headers set on every request to the
	// downstream, including any resolved connection-level secrets such as an
	// Authorization bearer token. The composition root resolves each declared
	// secret's env var to a value and passes it here, so the client never holds
	// the credential (ADR-0012). The gateway never reads these back to a client.
	Headers map[string]string
	// HTTPClient optionally overrides the HTTP client backing the transport (for
	// example to set timeouts or a custom dialer). When nil a zero-value client is
	// used. Configured Headers are injected regardless of which client backs it.
	HTTPClient *http.Client
}

HTTPConfig configures a remote Streamable HTTP downstream transport.

type HTTPFactory ¶

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

HTTPFactory creates remote-HTTP DownstreamSessions and satisfies the pool's Factory interface, so a supervised pool of remote sessions is just NewPool(NewHTTPFactory(cfg), poolCfg) — the same composition as stdio.

func NewHTTPFactory(cfg HTTPConfig) *HTTPFactory

func (f *HTTPFactory) New(ctx context.Context) (Session, error)

type ManagedDownstream ¶

type ManagedDownstream struct {
	Name string
	Mode SessionMode
	Pool *Pool
}

ManagedDownstream registers one downstream with the Manager: its name, its session-sharing mode, and the bounded pool that backs it. The composition root builds the pool from the downstream's supervised factory and pool config.

type Manager ¶

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

Manager owns the gateway's downstream pools and hands each client connection its own ClientSession. There is one Manager per gateway; it is safe for concurrent use. A background reaper closes held per_client sessions that have gone idle past their downstream's idle_ttl (design §4): such sessions stay in-use from the pool's view and so never face the pool's own idle reaper.

func NewManager(downstreams []ManagedDownstream) (*Manager, error)

func (m *Manager) Close() error

func (m *Manager) NewClientSession() *ClientSession

type Pool ¶

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

Pool is a bounded, idle-reaping pool of downstream sessions. At most Max sessions are live at once; an Acquire past Max waits up to AcquireTimeout then fails fast. It is safe for concurrent use.

The permit channel enforces the bound and gives Acquire a clean timeout; the mutex guards the idle set and counters. Idle sessions hold no permit, so the reaper can close them independently.

func NewPool(factory Factory, cfg PoolConfig) *Pool

func (p *Pool) Acquire(ctx context.Context) (Session, error)

func (p *Pool) Close() error

func (p *Pool) Discard(sess Session)

func (p *Pool) IdleTTL() time.Duration

func (p *Pool) Release(sess Session)

func (p *Pool) Stats() Stats

type PoolConfig ¶

type PoolConfig struct {
	// Max is the maximum number of live sessions.
	Max int
	// AcquireTimeout bounds how long Acquire waits for a slot before failing fast.
	AcquireTimeout time.Duration
	// IdleTTL is how long an idle session is kept before being reaped.
	IdleTTL time.Duration
}

PoolConfig bounds and tunes a pool. Zero fields fall back to package defaults.

type Session ¶

type Session interface {
	Close() error
}

Session is a poolable downstream session. The pool only needs to be able to close one; richer behavior (calling tools, pinging) is added by the concrete session types built on the pool.

type SessionMode ¶

type SessionMode int

SessionMode selects how a downstream's sessions are shared across client sessions (design §4, ADR-0002).

const (
	// PerClient gives each client session its own dedicated session to the
	// downstream. It is the default and is correct for stateful servers: a client
	// session reuses one held session across its calls.
	PerClient SessionMode = iota
	// Shared checks a session out of and back into the pool per call, so a
	// stateless downstream's sessions cycle across client sessions.
	Shared
)

type Stats ¶

type Stats struct {
	InUse   int
	Idle    int
	Created int64
}

Stats is a snapshot of a pool's occupancy.

type StdioConfig ¶

type StdioConfig struct {
	// Command is the downstream's argv; Command[0] is the executable.
	Command []string
	// Env are extra environment variables for the child, layered over the
	// gateway's own environment so PATH and friends still resolve. Includes
	// resolved connection-level secrets.
	Env map[string]string
	// TerminateDuration is how long Close waits after closing stdin before
	// escalating to SIGTERM. Zero uses the SDK default.
	TerminateDuration time.Duration
}

StdioConfig configures a downstream subprocess transport. Env is the full set of extra environment variables handed to the child, including any resolved connection-level secrets — the composition root resolves each declared secret's env var to a value and passes it here (ADR-0012).

type StdioFactory ¶

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

StdioFactory creates stdio DownstreamSessions and satisfies the pool's Factory interface, so a bounded pool of downstream subprocesses is just NewPool(NewStdioFactory(cfg), poolCfg).

func NewStdioFactory(cfg StdioConfig) *StdioFactory

func (f *StdioFactory) New(ctx context.Context) (Session, error)

type SupervisedFactory ¶

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

SupervisedFactory wraps a Factory with crash-loop storm prevention. While healthy it forwards every New to the delegate. After MaxConsecutiveFailures consecutive failures it trips a breaker: subsequent New calls fast-fail with ErrBroken without touching the delegate until BrokenCooldown elapses. The first New after the cooldown is a single probe that calls the delegate again; a successful probe clears the breaker, and a failed probe re-arms the cooldown.

It satisfies Factory, so it composes with the pool by wrapping any factory:

NewPool(NewSupervisedFactory(stdioFactory, supCfg), poolCfg)

It is safe for concurrent use; multiple pool Acquires call New concurrently.

func NewSupervisedFactory(delegate Factory, cfg SupervisorConfig) *SupervisedFactory

func (f *SupervisedFactory) New(ctx context.Context) (Session, error)

type SupervisorConfig ¶

type SupervisorConfig struct {
	// MaxConsecutiveFailures is how many consecutive New failures trip the breaker.
	// The failure that reaches this count is the one that marks the downstream
	// broken (it still returns the delegate's error; the next call fast-fails).
	MaxConsecutiveFailures int
	// BrokenCooldown is how long the breaker stays open before allowing a single
	// probe attempt.
	BrokenCooldown time.Duration
}

SupervisorConfig tunes a SupervisedFactory. A field that is <= 0 falls back to the corresponding package default.

type ToolInfo ¶

type ToolInfo struct {
	// Name is the downstream's own tool name (not yet namespaced).
	Name string
	// Title is the tool's optional human-readable display name.
	Title string
	// Description is the human-readable tool description, if any.
	Description string
	// InputSchema is the tool's JSON Schema as raw JSON, or nil if the downstream
	// advertised none. Kept as raw bytes so it round-trips unmodified.
	InputSchema json.RawMessage
	// OutputSchema is the tool's optional structured-result JSON Schema as raw
	// JSON, or nil. Clients validate structured results against it.
	OutputSchema json.RawMessage
	// Annotations is the tool's optional annotations (display and behavior hints)
	// as raw JSON, or nil. Kept opaque; the gateway does not interpret them.
	Annotations json.RawMessage
	// Icons is the tool's optional icon set as raw JSON, or nil.
	Icons json.RawMessage
}

ToolInfo is a downstream's raw, un-namespaced tool as discovered over the wire. The composition root maps these to the gateway's aggregate listing, applying the per-downstream namespace; this package stays transport-level and leaves namespacing and policy filtering to higher layers.