kami/Maven
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

mcp: guard the connection, not just the first dial

Browse Source 
Refresh called alive() with the manager lock held, so a slow health
check blocked every other server. It now snapshots the candidates and
asks outside the lock.

A server that cannot be dialled was retried every minute forever, which
for a misconfigured stdio block means re-exec'ing a process 1440 times a
day. Dials now back off from one minute to thirty.

An allow_private fetcher followed redirects. A LAN MCP endpoint could
answer a POST with a redirect to 169.254.169.254 and the guard would go
there, because allow_private is what turns the address check off.
Redirects are refused outright on that door.

The tool catalogue was trimmed by taking the first max_tools entries of
whatever order the server sent, so the server chose which of its tools
Maven proposed. Over the cap without allow_tools now contributes
nothing: refusing is honest, silently keeping the server's pick is not.
Descriptions are server-written text that lands in the router prompt and
on /tools, so they are capped too.

A server block with enabled false was skipped by validation, so a typo
in a block written dark surfaced only on the day it was switched on. All
blocks are shape-checked now. Configured static headers carry the bearer
token a real remote server needs, and host_interval bounds how fast one
endpoint is polled.

Found in review of #70.
This commit is contained in:
kami
2026-08-01 14:11:39 +04:00
parent 87d03cf8c6
commit 5e0417306b
5 changed files with 337 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/mcp/manager.go
+146 -27
View File
@@ -24,12 +24,31 @@ const (
// wrong act. Twelve per server is already generous.
DefaultMaxTools = 12
// DefaultReconnectEvery is how long the manager waits before re-dialing a
// server whose connection died.
// server whose connection died. It is the FIRST wait: every consecutive
// failure doubles it, up to MaxReconnectEvery.
DefaultReconnectEvery = 30 * time.Second
// MaxReconnectEvery caps the backoff. Without one, a permanently
// misconfigured stdio server is exec'd once a minute forever, which is a
// process spawn per minute in the logs and nothing that ever gets better.
MaxReconnectEvery = 30 * time.Minute
// DefaultMaxDescription bounds one tool description. It is written by a
// server Maven does not control and it lands in two places that cannot
// absorb an arbitrary blob: the resident model's 4096-token context, and a
// table cell on /tools.
DefaultMaxDescription = 400
)
// ErrNoServer — the named server is not configured or not connected.
var ErrNoServer = errors.New("mcp: no such server")
var (
// ErrNoServer — the named server is not configured.
ErrNoServer = errors.New("mcp: no such server")
// ErrNotConnected — the server is configured but nothing is dialed. Held
// apart from ErrNoServer so a caller can say "that tool is not connected"
// instead of drafting a proposal for a tool that already exists.
ErrNotConnected = errors.New("mcp: server is not connected")
// ErrToolGone — the server no longer offers this tool. An enabled row can
// outlive the tool it names; this is what the act path sees when it does.
ErrToolGone = errors.New("mcp: server no longer offers this tool")
)
// ServerConfig is one configured MCP server. Off unless present.
//
@@ -61,6 +80,10 @@ type ServerConfig struct {
AllowTools []string `json:"allow_tools,omitempty"`
// MaxTools caps the contribution (0 ⇒ DefaultMaxTools).
MaxTools int `json:"max_tools,omitempty"`
// Headers are sent verbatim on every request to a url server. This is how
// a bearer token reaches a real remote server; the Vikunja one on loopback
// needs none only because it is unauthenticated.
Headers map[string]string `json:"-"`
// Timeout bounds one call (0 ⇒ DefaultTimeout).
Timeout time.Duration `json:"-"`
// Enabled=false keeps a configured server described but dark.
@@ -90,6 +113,20 @@ type conn struct {
lastErr error
lastTry time.Time
dialedAt time.Time
fails int // consecutive dial failures, for the backoff
}
// backoff is how long this connection waits before the next dial attempt:
// DefaultReconnectEvery doubled per consecutive failure, capped.
func (c *conn) backoff() time.Duration {
d := DefaultReconnectEvery
for i := 1; i < c.fails && d < MaxReconnectEvery; i++ {
d *= 2
}
if d > MaxReconnectEvery {
d = MaxReconnectEvery
}
return d
}
// NewManager builds a manager for the enabled servers in cfgs. newPoster is
@@ -202,7 +239,7 @@ func (m *Manager) dial(ctx context.Context, name string) error {
} else {
var poster Poster
if poster, err = m.newPoster(cfg); err == nil {
tr = newHTTPTransport(poster, cfg.URL)
tr = newHTTPTransport(poster, cfg.URL, cfg.Headers)
}
}
if err != nil {
@@ -233,6 +270,7 @@ func (m *Manager) dial(ctx context.Context, name string) error {
m.conns[name].client = cl
m.conns[name].tools = tools
m.conns[name].lastErr = nil
m.conns[name].fails = 0
m.conns[name].dialedAt = time.Now()
m.mu.Unlock()
log.Printf("mcp: %s connected (%s %s), %d tool(s)", name, cl.Info().Name, cl.Info().Version, len(tools))
@@ -246,12 +284,19 @@ func (m *Manager) fail(name string, err error) {
c.lastErr = err
c.client = nil
c.tools = nil
c.fails++
}
}
// filterTools applies AllowTools and MaxTools, and drops nameless entries.
// Sorted first, so the cap is deterministic rather than "whatever order the
// server felt like".
// filterTools applies AllowTools and MaxTools, drops nameless entries and
// truncates descriptions.
//
// Over the cap WITHOUT allow_tools, the whole contribution is dropped. Taking
// the first N of a sorted list was deterministic but it handed the choice of
// which N to the server: a thirteenth tool named "aaa_" would push a tool that
// had already been discovered, proposed and maybe enabled out of the
// catalogue. Determinism was not the property worth buying. With allow_tools
// set, Kami named the tools, so the cap trims a list he chose.
func filterTools(cfg ServerConfig, in []Tool) []Tool {
sort.Slice(in, func(i, j int) bool { return in[i].Name < in[j].Name })
out := make([]Tool, 0, len(in))
@@ -259,16 +304,32 @@ func filterTools(cfg ServerConfig, in []Tool) []Tool {
if len(cfg.AllowTools) > 0 && !contains(cfg.AllowTools, t.Name) {
continue
}
t.Description = truncate(t.Description, DefaultMaxDescription)
out = append(out, t)
}
if cfg.MaxTools > 0 && len(out) > cfg.MaxTools {
log.Printf("mcp: %s offers %d tools, taking the first %d (raise max_tools or set allow_tools)",
cfg.Name, len(out), cfg.MaxTools)
if len(cfg.AllowTools) == 0 {
log.Printf("mcp: %s offers %d tools, over the cap of %d — taking NONE of them, set allow_tools to choose or raise max_tools",
cfg.Name, len(out), cfg.MaxTools)
return nil
}
log.Printf("mcp: %s: allow_tools names %d tools, over the cap of %d — taking the first %d",
cfg.Name, len(out), cfg.MaxTools, cfg.MaxTools)
out = out[:cfg.MaxTools]
}
return out
}
// truncate bounds a server-written string. The ellipsis is there so a reader
// on /tools can tell the text was cut rather than written that way.
func truncate(s string, max int) string {
r := []rune(s)
if len(r) <= max {
return s
}
return strings.TrimSpace(string(r[:max])) + "…"
}
func contains(hay []string, needle string) bool {
for _, h := range hay {
if h == needle {
@@ -281,18 +342,33 @@ func contains(hay []string, needle string) bool {
// Refresh re-dials any server that is down, if enough time has passed since the
// last attempt. Call it from the daemon's periodic tick — it is cheap when
// everything is up.
// The health check itself is done OUTSIDE m.mu, the way Resources already
// does it. alive() reaches into the transport, and a transport waiting on a
// silent subprocess would otherwise hold m.mu for as long as it waits, which
// blocks Tools, Status and Call for every other server too.
func (m *Manager) Refresh(ctx context.Context) {
now := time.Now()
var stale []string
type candidate struct {
name string
cl *Client
ready bool
}
var cands []candidate
m.mu.Lock()
for _, name := range m.order {
c := m.conns[name]
down := c.client == nil || !c.client.alive()
if down && now.Sub(c.lastTry) >= DefaultReconnectEvery {
stale = append(stale, name)
}
cands = append(cands, candidate{name: name, cl: c.client, ready: now.Sub(c.lastTry) >= c.backoff()})
}
m.mu.Unlock()
var stale []string
for _, c := range cands {
if !c.ready {
continue
}
if c.cl == nil || !c.cl.alive() {
stale = append(stale, c.name)
}
}
for _, name := range stale {
if err := m.dial(ctx, name); err != nil {
log.Printf("mcp: %s: reconnect: %v", name, err)
@@ -312,6 +388,21 @@ func (m *Manager) Tools() []Tool {
return out
}
// Connected — the names of servers that are dialed right now. A caller that
// wants to act on a tool's ABSENCE needs this: a tool missing from Tools()
// because its server is down is not a tool the server withdrew.
func (m *Manager) Connected() []string {
m.mu.Lock()
defer m.mu.Unlock()
var out []string
for _, name := range m.order {
if m.conns[name].client != nil {
out = append(out, name)
}
}
return out
}
// Status is one server's health, for the web surface.
type Status struct {
Name string
@@ -367,13 +458,13 @@ func (m *Manager) Call(ctx context.Context, server, tool string, args map[string
}
m.mu.Unlock()
if cl == nil {
return "", fmt.Errorf("mcp: %s is not connected", server)
return "", fmt.Errorf("%w: %s", ErrNotConnected, server)
}
// The discovered-and-filtered set is the second allowlist: even an enabled
// store row cannot reach a tool the server stopped offering, or one
// allow_tools excludes.
if !known {
return "", fmt.Errorf("mcp: %s offers no tool %q", server, tool)
return "", fmt.Errorf("%w: %s/%s", ErrToolGone, server, tool)
}
cctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
@@ -442,8 +533,8 @@ var ErrNeedsArgs = errors.New("mcp: tool needs named arguments")
//
// - a tool with no required properties runs with no arguments (a spare tail
// is ignored — "покажи проекты пожалуйста" should still list projects);
// - a READ-ONLY tool with exactly one required property, of type string or
// integer/number, gets the tail bound to it;
// - a READ-ONLY tool NAMED IN allow_tools, with exactly one required
// property, of type string or integer/number, gets the tail bound to it;
// - anything else is refused with ErrNeedsArgs. Such a tool is still callable
// with explicit arguments from the authed surface, where a human types
// them.
@@ -456,32 +547,54 @@ var ErrNeedsArgs = errors.New("mcp: tool needs named arguments")
// partially-filled write can do is destroy what it did not mention. A mutating
// tool with nothing required is still fine: nothing was guessed, and it still
// goes through the confirm turn.
//
// The allow_tools condition is the second half, and it is there because
// readOnlyHint is the SERVER's claim about itself. It already buys one
// exemption (destructive=false, so no confirm turn); letting it buy argument
// binding as well means one lie converts a spoken utterance into an
// unconfirmed, argument-carrying write. A server advertising delete_project
// with readOnlyHint true and the description "show a project and its tasks"
// would be enough. So the binding half rests on something local instead: a
// name Kami typed into mavend.json. The tool name is not a defence — the
// router picks tools by name similarity and the description a human reads is
// server-written too.
func (m *Manager) CallPositional(ctx context.Context, server, tool string, args []string) (string, error) {
m.mu.Lock()
c := m.conns[server]
var schema json.RawMessage
found, readOnly := false, false
found, readOnly, bindable := false, false, false
configured, connected := c != nil, false
if c != nil {
connected = c.client != nil
for _, t := range c.tools {
if t.Name == tool {
schema, readOnly, found = t.InputSchema, t.ReadOnly, true
bindable = contains(c.cfg.AllowTools, tool)
break
}
}
}
m.mu.Unlock()
if !found {
return "", fmt.Errorf("mcp: %s offers no tool %q", server, tool)
if !configured {
return "", fmt.Errorf("%w: %s", ErrNoServer, server)
}
if !connected {
return "", fmt.Errorf("%w: %s", ErrNotConnected, server)
}
return "", fmt.Errorf("%w: %s/%s", ErrToolGone, server, tool)
}
named, err := bindPositional(schema, args, readOnly)
named, err := bindPositional(schema, args, readOnly && bindable)
if err != nil {
return "", err
}
return m.Call(ctx, server, tool, named)
}
// bindPositional implements the rule documented on CallPositional.
func bindPositional(schema json.RawMessage, args []string, readOnly bool) (map[string]any, error) {
// bindPositional implements the rule documented on CallPositional. bind is the
// caller's verdict on whether a guessed argument is allowed at all: read-only
// AND named in allow_tools.
func bindPositional(schema json.RawMessage, args []string, bind bool) (map[string]any, error) {
var s struct {
Required []string `json:"required"`
Properties map[string]struct {
@@ -498,14 +611,20 @@ func bindPositional(schema json.RawMessage, args []string, readOnly bool) (map[s
return map[string]any{}, nil
case 1:
name := s.Required[0]
if !readOnly {
return nil, fmt.Errorf("%w: %q, and a tool that writes never gets a guessed one", ErrNeedsArgs, name)
prop, described := s.Properties[name]
if !described {
// required names it, properties does not describe it. The zero
// value would make it a string, which is a guess about a guess.
return nil, fmt.Errorf("%w: %q, which the schema never describes", ErrNeedsArgs, name)
}
if !bind {
return nil, fmt.Errorf("%w: %q, and a guessed argument goes only to a read-only tool named in allow_tools", ErrNeedsArgs, name)
}
tail := strings.TrimSpace(strings.Join(args, " "))
if tail == "" {
return nil, fmt.Errorf("%w: %q", ErrNeedsArgs, name)
}
switch s.Properties[name].Type {
switch prop.Type {
case "string", "":
return map[string]any{name: tail}, nil
case "integer", "number":
@@ -515,7 +634,7 @@ func bindPositional(schema json.RawMessage, args []string, readOnly bool) (map[s
}
return map[string]any{name: n}, nil
default:
return nil, fmt.Errorf("%w: %q is a %s", ErrNeedsArgs, name, s.Properties[name].Type)
return nil, fmt.Errorf("%w: %q is a %s", ErrNeedsArgs, name, prop.Type)
}
default:
return nil, fmt.Errorf("%w: %s", ErrNeedsArgs, strings.Join(s.Required, ", "))