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

mcp manager: one lookup for the call paths, and split dial (V-581)

Browse Source 
Call, ReadResource and CallPositional each opened the connection map by
hand, and Call took the mutex twice to answer one question. They now
share lookup, which returns the client, the config and the tool in one
critical section.

dial did three things. Choosing the transport is openTransport, and
recording a live connection is succeed, so the function reads as
handshake then discovery.

Also: argv is a method rather than an append repeated in dial and
Status, the transport strings are constants, and the tail binding comes
out of bindPositional as bindOne. The comment on filterTools claimed it
drops nameless tools, which it never did.

No behaviour change.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
claude
2026-08-06 01:57:27 +04:00
parent ecaf3407fa
commit 439ceb5d8e
1 changed files with 102 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/mcp/manager.go
+102 -85
View File
@@ -38,6 +38,12 @@ const (
DefaultMaxDescription = 400
)
// The two transports, as Status reports them to the web surface.
const (
transportStdio = "stdio"
transportHTTP = "http"
)
var (
// ErrNoServer — the named server is not configured.
ErrNoServer = errors.New("mcp: no such server")
@@ -90,6 +96,12 @@ type ServerConfig struct {
Enabled bool `json:"enabled"`
}
// argv is the stdio server's command line. It is argv and never a shell
// string, so the same slice serves both the exec and the /tools display.
func (c ServerConfig) argv() []string {
return append([]string{c.Command}, c.Args...)
}
// PosterFactory builds the HTTP door for one server. It is a factory rather
// than a single shared Poster because allow_private is per server: the fetcher
// that may reach http://localhost:9100/mcp must NOT be the same fetcher another
@@ -230,18 +242,7 @@ func (m *Manager) dial(ctx context.Context, name string) error {
c.lastTry = time.Now()
m.mu.Unlock()
var tr transport
var err error
if cfg.Command != "" {
tr, err = newStdioTransport(ctx, append([]string{cfg.Command}, cfg.Args...), cfg.Env, cfg.Dir)
} else if m.newPoster == nil {
err = fmt.Errorf("server %q has a url but no http door was wired", name)
} else {
var poster Poster
if poster, err = m.newPoster(cfg); err == nil {
tr = newHTTPTransport(poster, cfg.URL, cfg.Headers)
}
}
tr, err := m.openTransport(ctx, cfg)
if err != nil {
m.fail(name, err)
return err
@@ -262,21 +263,44 @@ func (m *Manager) dial(ctx context.Context, name string) error {
tools = nil
}
tools = filterTools(cfg, tools)
m.mu.Lock()
if old := m.conns[name].client; old != nil {
_ = old.Close()
}
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()
m.succeed(name, cl, tools)
log.Printf("mcp: %s connected (%s %s), %d tool(s)", name, cl.Info().Name, cl.Info().Version, len(tools))
return nil
}
// openTransport builds the door this server is configured for. A url server
// with no factory is one server's problem, reported here, so a bad block never
// stops the daemon.
func (m *Manager) openTransport(ctx context.Context, cfg ServerConfig) (transport, error) {
if cfg.Command != "" {
return newStdioTransport(ctx, cfg.argv(), cfg.Env, cfg.Dir)
}
if m.newPoster == nil {
return nil, fmt.Errorf("server %q has a url but no http door was wired", cfg.Name)
}
poster, err := m.newPoster(cfg)
if err != nil {
return nil, err
}
return newHTTPTransport(poster, cfg.URL, cfg.Headers), nil
}
// succeed records a live connection and closes the one it replaces, so a
// re-dial does not leak the previous subprocess.
func (m *Manager) succeed(name string, cl *Client, tools []Tool) {
m.mu.Lock()
defer m.mu.Unlock()
c := m.conns[name]
if c == nil {
return
}
if c.client != nil {
_ = c.client.Close()
}
c.client, c.tools, c.lastErr, c.fails = cl, tools, nil, 0
c.dialedAt = time.Now()
}
func (m *Manager) fail(name string, err error) {
m.mu.Lock()
defer m.mu.Unlock()
@@ -288,8 +312,7 @@ func (m *Manager) fail(name string, err error) {
}
}
// filterTools applies AllowTools and MaxTools, drops nameless entries and
// truncates descriptions.
// filterTools applies AllowTools and MaxTools 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
@@ -423,9 +446,9 @@ func (m *Manager) Status() []Status {
c := m.conns[name]
s := Status{Name: name, Tools: len(c.tools)}
if c.cfg.Command != "" {
s.Transport, s.Target = "stdio", strings.Join(append([]string{c.cfg.Command}, c.cfg.Args...), " ")
s.Transport, s.Target = transportStdio, strings.Join(c.cfg.argv(), " ")
} else {
s.Transport, s.Target = "http", c.cfg.URL
s.Transport, s.Target = transportHTTP, c.cfg.URL
}
if c.client != nil {
s.Connected = true
@@ -442,21 +465,10 @@ func (m *Manager) Status() []Status {
// Call runs server's tool with args. Args come from the router and nothing
// else; there is no path here through which a note or a fact could travel.
func (m *Manager) Call(ctx context.Context, server, tool string, args map[string]any) (string, error) {
m.mu.Lock()
c := m.conns[server]
m.mu.Unlock()
if c == nil {
cl, cfg, _, configured, known := m.lookup(server, tool)
if !configured {
return "", fmt.Errorf("%w: %s", ErrNoServer, server)
}
m.mu.Lock()
cl, timeout, known := c.client, c.cfg.Timeout, false
for _, t := range c.tools {
if t.Name == tool {
known = true
break
}
}
m.mu.Unlock()
if cl == nil {
return "", fmt.Errorf("%w: %s", ErrNotConnected, server)
}
@@ -466,11 +478,30 @@ func (m *Manager) Call(ctx context.Context, server, tool string, args map[string
if !known {
return "", fmt.Errorf("%w: %s/%s", ErrToolGone, server, tool)
}
cctx, cancel := context.WithTimeout(ctx, timeout)
cctx, cancel := context.WithTimeout(ctx, cfg.Timeout)
defer cancel()
return cl.CallTool(cctx, tool, args)
}
// lookup reads one server's live state under the lock. Every call path needs
// the same four answers, and reading them in one critical section keeps a
// server that goes down mid-check from answering half yes.
func (m *Manager) lookup(server, tool string) (cl *Client, cfg ServerConfig, found Tool, configured, known bool) {
m.mu.Lock()
defer m.mu.Unlock()
c := m.conns[server]
if c == nil {
return nil, ServerConfig{}, Tool{}, false, false
}
for _, t := range c.tools {
if t.Name == tool {
found, known = t, true
break
}
}
return c.client, c.cfg, found, true, known
}
// Resources lists resources across connected servers.
func (m *Manager) Resources(ctx context.Context) []Resource {
m.mu.Lock()
@@ -494,18 +525,11 @@ func (m *Manager) Resources(ctx context.Context) []Resource {
// ReadResource reads one resource from one server.
func (m *Manager) ReadResource(ctx context.Context, server, uri string) (string, error) {
m.mu.Lock()
c := m.conns[server]
var cl *Client
var timeout time.Duration
if c != nil {
cl, timeout = c.client, c.cfg.Timeout
}
m.mu.Unlock()
cl, cfg, _, _, _ := m.lookup(server, "")
if cl == nil {
return "", fmt.Errorf("%w: %s", ErrNoServer, server)
}
cctx, cancel := context.WithTimeout(ctx, timeout)
cctx, cancel := context.WithTimeout(ctx, cfg.Timeout)
defer cancel()
return cl.ReadResource(cctx, uri)
}
@@ -559,32 +583,18 @@ var ErrNeedsArgs = errors.New("mcp: tool needs named arguments")
// 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, 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 {
cl, cfg, t, configured, known := m.lookup(server, tool)
if !known {
if !configured {
return "", fmt.Errorf("%w: %s", ErrNoServer, server)
}
if !connected {
if cl == nil {
return "", fmt.Errorf("%w: %s", ErrNotConnected, server)
}
return "", fmt.Errorf("%w: %s/%s", ErrToolGone, server, tool)
}
named, err := bindPositional(schema, args, readOnly && bindable)
bindable := t.ReadOnly && contains(cfg.AllowTools, tool)
named, err := bindPositional(t.InputSchema, args, bindable)
if err != nil {
return "", err
}
@@ -620,23 +630,30 @@ func bindPositional(schema json.RawMessage, args []string, bind bool) (map[strin
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 prop.Type {
case "string", "":
return map[string]any{name: tail}, nil
case "integer", "number":
n, err := strconv.ParseFloat(tail, 64)
if err != nil {
return nil, fmt.Errorf("%w: %q wants a number, got %q", ErrNeedsArgs, name, tail)
}
return map[string]any{name: n}, nil
default:
return nil, fmt.Errorf("%w: %q is a %s", ErrNeedsArgs, name, prop.Type)
}
return bindOne(name, prop.Type, args)
default:
return nil, fmt.Errorf("%w: %s", ErrNeedsArgs, strings.Join(s.Required, ", "))
}
}
// bindOne puts the whole positional tail in the one required property. The
// tail is spoken words, so only a scalar can hold it and anything else is
// refused rather than coerced.
func bindOne(name, typ string, args []string) (map[string]any, error) {
tail := strings.TrimSpace(strings.Join(args, " "))
if tail == "" {
return nil, fmt.Errorf("%w: %q", ErrNeedsArgs, name)
}
switch typ {
case "string", "":
return map[string]any{name: tail}, nil
case "integer", "number":
n, err := strconv.ParseFloat(tail, 64)
if err != nil {
return nil, fmt.Errorf("%w: %q wants a number, got %q", ErrNeedsArgs, name, tail)
}
return map[string]any{name: n}, nil
default:
return nil, fmt.Errorf("%w: %q is a %s", ErrNeedsArgs, name, typ)
}
}