diff --git a/README.md b/README.md index 98b9b2c..4d53a08 100644 --- a/README.md +++ b/README.md @@ -200,11 +200,11 @@ mcp_servers: # omitted by default; each entry runs a trusted loc ### Local MCP tools -MCP is disabled unless `mcp_servers` is explicitly configured. Each configured executable is trusted code with its own host and network permissions; this is tool preauthorization, not a sandbox or an interactive approval prompt. Configuration is read only from the user config or the explicitly selected `-c` file, never from the repository or model output. Tinyagent launches the command directly (no shell), with `--cwd` as its working directory and a minimal environment (`PATH`, plus the Windows process variables when present) and only the named forwarded variables. Missing forwarded variables fail before launch. Do not forward secrets unless the server needs them. +MCP is disabled unless `mcp_servers` is explicitly configured. Each configured executable is trusted code with its own host and network permissions; this is tool preauthorization, not a sandbox or an interactive approval prompt. Configuration is read only from the user config or the explicitly selected `-c` file, never from the repository or model output. Tinyagent launches the command directly (no shell), with `--cwd` as its working directory and a minimal environment (`PATH`, plus the Windows process variables when present) and only the named forwarded variables. Missing forwarded variables fail before launch. Do not forward secrets unless the server needs them. Read-only runs never start MCP servers. -The Go MCP SDK v1.8.0 supports protocol revisions 2026-07-28, 2025-11-25, 2025-06-18, 2025-03-26, and 2024-11-05 over stdio. Tinyagent discovers the complete paginated catalog once per run, with a 15-second default discovery timeout, 100-page, 256-selected-tool, 64-KiB-per-schema, and 1-MiB-total-schema caps, and exposes only configured names. Tool names are sent to providers as `__` and must fit provider-compatible naming rules. The accepted input schema subset supports object, array, string, number, integer, boolean and null types, nested properties/items, required, enum, simple bounds/length fields, and boolean `additionalProperties`; unsupported keywords and remote references are rejected. Each call defaults to a 60-second timeout. Text content is joined with newlines, structured output is appended as JSON, unsupported content is marked as omitted, tool errors remain visible, and model-visible output is capped at 256 KiB before the normal output cap is applied. Stdio protocol data is kept separate from server stderr. +The Go MCP SDK v1.8.0 supports protocol revisions 2026-07-28, 2025-11-25, 2025-06-18, 2025-03-26, and 2024-11-05 over stdio. Tinyagent discovers the complete paginated catalog once per run, with a 15-second default discovery timeout, 100-page, 256-selected-tool, 64-KiB-per-schema, and 1-MiB-total-schema caps, and exposes only configured names. Tool names are sent to providers as `__` and must fit provider-compatible naming rules. The accepted input schema subset supports object, array, string, number, integer, boolean and null types (a single type or an array such as `["string", "null"]`), nested properties/items, required, enum, const, numeric bounds and `multipleOf`, length, `pattern`, `format`, item and property counts, `uniqueItems`, annotations (`title`, `description`, `default`, `examples`), and boolean `additionalProperties`. A top-level `$schema` declaration is dropped. Other keywords, including `anyOf`/`oneOf`/`allOf` and references, are rejected, and a selected tool with such a schema stops startup. Each call defaults to a 60-second timeout; a timed-out call returns an error to the model and the server stays available, while any other call failure disables that server for the rest of the run. Text content is joined with newlines, structured output is appended as JSON, unsupported content is marked as omitted, tool errors remain visible, and model-visible output is capped at 256 KiB before the normal output cap is applied. Stdio protocol data is kept separate from server stderr. -MCP tool calls and results are stored in trajectories along with the external tool names. Trajectories from MCP-enabled runs cannot be resumed, and MCP configuration is rejected when resuming a built-in-only checkpoint. Trajectories and server stderr may contain sensitive data supplied by tools or the server; no blanket redaction is provided. On shutdown tinyagent closes MCP sessions and asks their processes to exit, then the SDK applies its bounded termination. Unix process descendants and Windows child-process cleanup have platform-specific behavior; tinyagent does not promise process-tree cleanup on Windows. +MCP tool calls and results are stored in trajectories along with the external tool names. Trajectories from MCP-enabled runs cannot be resumed. Resuming a built-in-only checkpoint works with `mcp_servers` configured; the servers are not started and the run keeps the built-in tools. Trajectories and server stderr may contain sensitive data supplied by tools or the server; no blanket redaction is provided. On shutdown tinyagent closes MCP sessions and asks their processes to exit, then the SDK applies its bounded termination. Unix process descendants and Windows child-process cleanup have platform-specific behavior; tinyagent does not promise process-tree cleanup on Windows. ## Long runs diff --git a/cmd/tinyagent/main.go b/cmd/tinyagent/main.go index b54f907..52a4ec8 100644 --- a/cmd/tinyagent/main.go +++ b/cmd/tinyagent/main.go @@ -157,13 +157,14 @@ func run() int { } ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) defer stop() - started := time.Now() + // MCP discovery counts against the wall-clock limit. Agent.Run applies + // the same limit again, and the earlier deadline set here wins. if cfg.WallTimeSec > 0 { var cancel context.CancelFunc ctx, cancel = context.WithTimeout(ctx, time.Duration(cfg.WallTimeSec)*time.Second) defer cancel() } - mcpRegistry, err := mcptools.Start(ctx, root, cfg.MCPServers, os.Stderr) + mcpRegistry, err := mcptools.Start(ctx, root, mcpServers(os.Stderr, cfg), os.Stderr) if err != nil { return fail(err) } @@ -202,19 +203,7 @@ func run() int { SkipAgentsMD: o.noAgentsMD, NoCompact: o.noCompact, } - events := newJSONEvents(os.Stdout, o.jsonOutput) - if o.jsonOutput { - events.emit("run.started", map[string]any{"schema_version": 1, "model": o.modelName, "root": root, "read_only": cfg.ReadOnly, "resumed": false}) - a.OnEvent = func(kind string, fields map[string]any) { - events.emit(kind, fields) - if events.err != nil { - stop() - } - } - if cfg.CostLimit > 0 && model.Unpriced(o.modelName) { - events.emit("warning", map[string]any{"message": fmt.Sprintf("no pricing available for %s; --cost-limit cannot be enforced", o.modelName)}) - } - } + events := startEvents(os.Stdout, o.jsonOutput, a, false, stop) if events.err != nil { return 1 } @@ -230,29 +219,12 @@ func run() int { } } - if cfg.WallTimeSec > 0 { - remaining := time.Duration(cfg.WallTimeSec)*time.Second - time.Since(started) - if remaining <= 0 { - return fail(errors.New("wall-clock time limit expired during MCP discovery")) - } - cfg.WallTimeSec = int(remaining / time.Second) - if cfg.WallTimeSec == 0 { - cfg.WallTimeSec = 1 - } - a.Config.WallTimeSec = cfg.WallTimeSec - } - runStart := time.Now() + start := time.Now() t := a.Run(ctx, task) - elapsed := time.Since(runStart) + elapsed := time.Since(start) reportSummary(os.Stderr, o.modelName, t, elapsed) - if o.jsonOutput { - if t.ExitStatus == agent.StatusError { - events.emit("error", map[string]any{"message": t.Submission}) - } - events.emit("result", resultFields(t, elapsed)) - if events.err != nil { - return 1 - } + if events.finish(t, elapsed) != nil { + return 1 } if t.ExitStatus == agent.StatusSubmitted || t.ExitStatus == agent.StatusCompleted { return 0 @@ -289,8 +261,11 @@ func runResume(o *options, fs *flag.FlagSet) int { if err != nil { return fail(err) } - if len(cfg.MCPServers) > 0 || src.MCPEnabled || len(src.ExternalTools) > 0 { - return fail(errors.New("resume: MCP-enabled runs cannot be resumed, and MCP configuration cannot be added to a built-in checkpoint; start a fresh task")) + if src.MCPEnabled || len(src.ExternalTools) > 0 { + return fail(errors.New("resume: MCP-enabled runs cannot be resumed; start a fresh task")) + } + if len(cfg.MCPServers) > 0 { + fmt.Fprintln(os.Stderr, "resume: configured MCP servers are not started; resumed runs use only the built-in tools") } applyFlagOverrides(fs, &cfg, o.stepLimit, o.costLimit, o.timeLimit) if flagWasSet(fs, "read-only") && o.readOnly != src.ReadOnly { @@ -374,19 +349,7 @@ func runResume(o *options, fs *flag.FlagSet) int { Resume: src, ResumeSource: o.resume, } - events := newJSONEvents(os.Stdout, o.jsonOutput) - if o.jsonOutput { - events.emit("run.started", map[string]any{"schema_version": 1, "model": modelName, "root": root, "read_only": src.ReadOnly, "resumed": true}) - a.OnEvent = func(kind string, fields map[string]any) { - events.emit(kind, fields) - if events.err != nil { - stop() - } - } - if cfg.CostLimit > 0 && model.Unpriced(modelName) { - events.emit("warning", map[string]any{"message": fmt.Sprintf("no pricing available for %s; --cost-limit cannot be enforced", modelName)}) - } - } + events := startEvents(os.Stdout, o.jsonOutput, a, true, stop) if events.err != nil { return 1 } @@ -405,18 +368,12 @@ func runResume(o *options, fs *flag.FlagSet) int { fmt.Fprintf(os.Stderr, "resuming %s from %s (previous status %s after %d steps)\n", modelName, o.resume, prevStatus, src.Steps) fmt.Fprintln(os.Stderr, "note: inspect and revalidate the workspace; it may have changed and commands may have produced unrecorded side effects after the last save. Live processes are not restored") - started := time.Now() + start := time.Now() t := a.Run(ctx, src.Task) - elapsed := time.Since(started) + elapsed := time.Since(start) reportSummary(os.Stderr, modelName, t, elapsed) - if o.jsonOutput { - if t.ExitStatus == agent.StatusError { - events.emit("error", map[string]any{"message": t.Submission}) - } - events.emit("result", resultFields(t, elapsed)) - if events.err != nil { - return 1 - } + if events.finish(t, elapsed) != nil { + return 1 } if t.ExitStatus == agent.StatusSubmitted || t.ExitStatus == agent.StatusCompleted { return 0 @@ -711,6 +668,47 @@ func (e *jsonEvents) emit(kind string, fields map[string]any) { e.err = err } +// startEvents writes run.started (and the unpriced cost-limit warning) and +// routes a's events to out. A failed write cancels the run through stop. +// When disabled it returns a no-op stream. +func startEvents(out io.Writer, enabled bool, a *agent.Agent, resumed bool, stop func()) *jsonEvents { + events := newJSONEvents(out, enabled) + if !enabled { + return events + } + events.emit("run.started", map[string]any{"schema_version": 1, "model": a.ModelName, "root": a.Tools.Root, "read_only": a.Tools.ReadOnly, "resumed": resumed}) + a.OnEvent = func(kind string, fields map[string]any) { + events.emit(kind, fields) + if events.err != nil { + stop() + } + } + if a.Config.CostLimit > 0 && model.Unpriced(a.ModelName) { + events.emit("warning", map[string]any{"message": fmt.Sprintf("no pricing available for %s; --cost-limit cannot be enforced", a.ModelName)}) + } + return events +} + +// finish writes the closing events for t and returns the first write error. +func (e *jsonEvents) finish(t *trajectory.Trajectory, elapsed time.Duration) error { + if t.ExitStatus == agent.StatusError { + e.emit("error", map[string]any{"message": t.Submission}) + } + e.emit("result", resultFields(t, elapsed)) + return e.err +} + +// mcpServers returns the MCP servers to start. Read-only runs start none: +// a server is an arbitrary local executable, and its tools are never offered +// there anyway. +func mcpServers(w io.Writer, cfg config.AgentConfig) []config.MCPServer { + if cfg.ReadOnly && len(cfg.MCPServers) > 0 { + fmt.Fprintln(w, "read-only mode: configured MCP servers are not started") + return nil + } + return cfg.MCPServers +} + func runOutput(jsonMode bool) io.Writer { if jsonMode { return io.Discard diff --git a/cmd/tinyagent/main_test.go b/cmd/tinyagent/main_test.go index 2cf861e..609c8a4 100644 --- a/cmd/tinyagent/main_test.go +++ b/cmd/tinyagent/main_test.go @@ -1,6 +1,7 @@ package main import ( + "context" "encoding/json" "errors" "flag" @@ -17,6 +18,7 @@ import ( "github.com/vicgupta/tinyagent/internal/agent" "github.com/vicgupta/tinyagent/internal/config" "github.com/vicgupta/tinyagent/internal/model" + "github.com/vicgupta/tinyagent/internal/tools" "github.com/vicgupta/tinyagent/internal/trajectory" ) @@ -162,6 +164,91 @@ func TestJSONEventsWriteOneJSONLine(t *testing.T) { } } +// replies is a model that returns its messages in order. +type replies []model.Message + +func (r *replies) Query(context.Context, []model.Message, []model.ToolSpec) (model.Message, model.Cost, error) { + if len(*r) == 0 { + return model.Message{}, model.Cost{}, errors.New("out of replies") + } + msg := (*r)[0] + *r = (*r)[1:] + return msg, model.Cost{InputTokens: 10, OutputTokens: 5}, nil +} + +func newTestAgent(t *testing.T, m model.Model) *agent.Agent { + t.Helper() + root := t.TempDir() + if err := os.Mkdir(filepath.Join(root, ".git"), 0o755); err != nil { + t.Fatal(err) + } + return &agent.Agent{Model: m, ModelName: "fake", Tools: &tools.Executor{Root: root}, Config: config.Default(), SkipAgentsMD: true} +} + +// TestJSONRunEventStream drives a whole run through startEvents and finish: +// stdout must be only JSONL, opening with run.started and closing with result. +func TestJSONRunEventStream(t *testing.T) { + m := &replies{ + {Role: "assistant", Content: "looking", ToolCalls: []model.ToolCall{{ID: "1", Type: "function", Function: model.FunctionCall{Name: "list_files", Arguments: `{}`}}}}, + {Role: "assistant", Content: "all good\n" + agent.SubmitMarker}, + } + a := newTestAgent(t, m) + var out strings.Builder + events := startEvents(&out, true, a, false, func() { t.Error("stop called on a healthy stream") }) + tr := a.Run(context.Background(), "check") + if err := events.finish(tr, time.Second); err != nil { + t.Fatal(err) + } + var kinds []string + var last map[string]any + for i, line := range strings.Split(strings.TrimSuffix(out.String(), "\n"), "\n") { + if err := json.Unmarshal([]byte(line), &last); err != nil { + t.Fatalf("line %d is not JSON: %q", i, line) + } + kinds = append(kinds, last["type"].(string)) + } + want := []string{"run.started", "assistant.message", "tool.call", "tool.result", "assistant.message", "result"} + if strings.Join(kinds, " ") != strings.Join(want, " ") { + t.Fatalf("events = %v, want %v", kinds, want) + } + if last["status"] != agent.StatusSubmitted || last["submission"] != "all good" || last["duration_ms"] != float64(1000) { + t.Errorf("result = %#v", last) + } +} + +func TestJSONEventsDisabledWriteNothing(t *testing.T) { + a := newTestAgent(t, &replies{{Role: "assistant", Content: "done"}}) + var out strings.Builder + events := startEvents(&out, false, a, false, func() {}) + if err := events.finish(a.Run(context.Background(), "x"), 0); err != nil || out.Len() != 0 || a.OnEvent != nil { + t.Errorf("disabled stream wrote %q (err %v, OnEvent set %v)", out.String(), err, a.OnEvent != nil) + } +} + +func TestReadOnlyStartsNoMCPServers(t *testing.T) { + servers := []config.MCPServer{{Alias: "docs", Command: "docs-mcp", Tools: []string{"search"}}} + var w strings.Builder + if got := mcpServers(&w, config.AgentConfig{MCPServers: servers}); len(got) != 1 || w.Len() != 0 { + t.Errorf("normal mode: servers %v, note %q", got, w.String()) + } + if got := mcpServers(&w, config.AgentConfig{ReadOnly: true, MCPServers: servers}); got != nil || !strings.Contains(w.String(), "not started") { + t.Errorf("read-only mode: servers %v, note %q", got, w.String()) + } +} + +// TestDeadlineSpentBeforeRunIsTimeLimit covers a wall-clock limit used up by +// MCP discovery: the run still ends as TimeLimitExceeded. +func TestDeadlineSpentBeforeRunIsTimeLimit(t *testing.T) { + a := newTestAgent(t, &replies{{Role: "assistant", Content: "done"}}) + a.Config.WallTimeSec = 60 + ctx, cancel := context.WithTimeout(context.Background(), time.Nanosecond) + defer cancel() + <-ctx.Done() + if tr := a.Run(ctx, "x"); tr.ExitStatus != agent.StatusTimeLimit { + t.Errorf("status = %s, want %s", tr.ExitStatus, agent.StatusTimeLimit) + } +} + // wantUsage is the layout from issue #12, with its update: -h, --help listed // first and -V for --version. const wantUsage = `Usage: tinyagent -t "" [flags] (or --task "") diff --git a/internal/agent/loop_test.go b/internal/agent/loop_test.go index 80d3f10..36c5028 100644 --- a/internal/agent/loop_test.go +++ b/internal/agent/loop_test.go @@ -181,14 +181,14 @@ func TestRunDispatchesExternalToolsWithEvents(t *testing.T) { }} a := newAgent(t, m) a.Tools.ExternalSpecs = []model.ToolSpec{{Type: "function", Function: model.FunctionSpec{Name: "docs__search"}}} - a.Tools.ExternalCall = func(_ context.Context, name, raw string) (string, bool) { + a.Tools.ExternalCall = func(_ context.Context, name, raw string) (string, bool, bool) { if name != "docs__search" { - return "", false + return "", false, false } if raw == `{}` { - return "tool error: missing q", true + return "tool error: missing q", true, true } - return "result for " + raw, true + return "result for " + raw, false, true } results := map[string]map[string]any{} a.OnEvent = func(kind string, fields map[string]any) { diff --git a/internal/mcptools/client.go b/internal/mcptools/client.go index 3fa20fc..cd98e90 100644 --- a/internal/mcptools/client.go +++ b/internal/mcptools/client.go @@ -63,153 +63,124 @@ func Start(ctx context.Context, root string, cfg []config.MCPServer, stderr io.W if err := validateDefinitions(cfg); err != nil { return nil, err } - reserved := map[string]bool{} - for _, s := range []string{"read_file", "write_file", "edit_file", "list_files", "grep", "bash", "start_process", "read_process", "stop_process"} { - reserved[s] = true - } - aliases := map[string]bool{} schemaBytes := 0 - fail := func(err error) (*Registry, error) { r.Close(); return nil, err } for _, def := range cfg { - if !validName(def.Alias) || aliases[def.Alias] { - return fail(fmt.Errorf("mcp server %q: alias must be unique and contain only letters, digits, _ or -", def.Alias)) - } - aliases[def.Alias] = true - if def.Command == "" || len(def.Tools) == 0 { - return fail(fmt.Errorf("mcp server %q: command and a nonempty tools allowlist are required", def.Alias)) - } - if def.DiscoveryTimeoutSec == 0 { - def.DiscoveryTimeoutSec = 15 + if err := r.start(ctx, root, def, stderr, &schemaBytes); err != nil { + r.Close() + return nil, err } - if def.CallTimeoutSec == 0 { - def.CallTimeoutSec = 60 - } - if def.DiscoveryTimeoutSec < 1 || def.DiscoveryTimeoutSec > 300 || def.CallTimeoutSec < 1 || def.CallTimeoutSec > 600 { - return fail(fmt.Errorf("mcp server %q: timeouts must be between 1 and 300 seconds (discovery) and 1 and 600 seconds (call)", def.Alias)) - } - env := make([]string, 0, len(def.Env)+2) - if p := os.Getenv("PATH"); p != "" { - env = append(env, "PATH="+p) - } - if runtime.GOOS == "windows" { - for _, k := range []string{"SystemRoot", "WINDIR", "USERPROFILE", "TEMP", "TMP"} { - if v, ok := os.LookupEnv(k); ok { - env = append(env, k+"="+v) - } + fmt.Fprintf(stderr, "mcp: enabled server %s (%d selected tools)\n", def.Alias, len(def.Tools)) + } + if len(r.Specs) == 0 { + r.Close() + return nil, errors.New("mcp: configured servers produced no tools") + } + return r, nil +} + +// start launches one validated server and adds its allowlisted tools to the +// catalog. schemaBytes accumulates the catalog's schema size across servers. +func (r *Registry) start(ctx context.Context, root string, def config.MCPServer, stderr io.Writer, schemaBytes *int) error { + if def.DiscoveryTimeoutSec == 0 { + def.DiscoveryTimeoutSec = 15 + } + if def.CallTimeoutSec == 0 { + def.CallTimeoutSec = 60 + } + env := make([]string, 0, len(def.Env)+2) + if p := os.Getenv("PATH"); p != "" { + env = append(env, "PATH="+p) + } + if runtime.GOOS == "windows" { + for _, k := range []string{"SystemRoot", "WINDIR", "USERPROFILE", "TEMP", "TMP"} { + if v, ok := os.LookupEnv(k); ok { + env = append(env, k+"="+v) } } - seenEnv := map[string]bool{} - for _, k := range def.Env { - if k == "" || strings.ContainsAny(k, "=\x00") || seenEnv[k] { - return fail(fmt.Errorf("mcp server %q: invalid or duplicate forwarded environment variable name", def.Alias)) - } - seenEnv[k] = true - v, ok := os.LookupEnv(k) - if !ok { - return fail(fmt.Errorf("mcp server %q: forwarded environment variable %s is not set", def.Alias, k)) - } - env = append(env, k+"="+v) + } + for _, k := range def.Env { + env = append(env, k+"="+os.Getenv(k)) // presence checked by validateDefinitions + } + cmd := exec.Command(def.Command, def.Args...) + cmd.Dir = root + cmd.Env = env + cmd.Stderr = &limitedWriter{w: stderr, n: 32 << 10} + client := mcp.NewClient(&mcp.Implementation{Name: "tinyagent", Version: "1"}, nil) + dctx, cancel := context.WithTimeout(ctx, time.Duration(def.DiscoveryTimeoutSec)*time.Second) + defer cancel() + sess, err := client.Connect(dctx, &mcp.CommandTransport{Command: cmd, TerminateDuration: 2 * time.Second}, nil) + if err != nil { + return fmt.Errorf("mcp server %q: connect failed: %w", def.Alias, err) + } + r.sessions = append(r.sessions, sess) + sv := &server{alias: def.Alias, session: sess, timeout: time.Duration(def.CallTimeoutSec) * time.Second} + selected := map[string]bool{} + for _, name := range def.Tools { + selected[name] = true + } + found := map[string]bool{} + cursor := "" + cursors := map[string]bool{} + count := 0 + for page := 0; ; page++ { + if page == maxPages { + return fmt.Errorf("mcp server %q: discovery exceeds %d pages", def.Alias, maxPages) } - cmd := exec.Command(def.Command, def.Args...) - cmd.Dir = root - cmd.Env = env - cmd.Stderr = &limitedWriter{w: stderr, n: 32 << 10} - client := mcp.NewClient(&mcp.Implementation{Name: "tinyagent", Version: "1"}, nil) - dctx, cancel := context.WithTimeout(ctx, time.Duration(def.DiscoveryTimeoutSec)*time.Second) - defer cancel() - sess, err := client.Connect(dctx, &mcp.CommandTransport{Command: cmd, TerminateDuration: 2 * time.Second}, nil) + res, err := sess.ListTools(dctx, &mcp.ListToolsParams{Cursor: cursor}) if err != nil { - return fail(fmt.Errorf("mcp server %q: connect failed: %w", def.Alias, err)) + return fmt.Errorf("mcp server %q: tool discovery failed: %w", def.Alias, err) } - r.sessions = append(r.sessions, sess) - sv := &server{alias: def.Alias, session: sess, timeout: time.Duration(def.CallTimeoutSec) * time.Second} - selected := map[string]bool{} - for _, name := range def.Tools { - if name == "" || selected[name] { - return fail(fmt.Errorf("mcp server %q: tool allowlist contains an empty or duplicate name", def.Alias)) - } - selected[name] = true + count += len(res.Tools) + if count > maxTools { + return fmt.Errorf("mcp server %q: discovery exceeds %d tools", def.Alias, maxTools) } - found := map[string]bool{} - cursor := "" - cursors := map[string]bool{} - count := 0 - for page := 0; page < maxPages; page++ { - res, err := sess.ListTools(dctx, &mcp.ListToolsParams{Cursor: cursor}) - if err != nil { - cancel() - return fail(fmt.Errorf("mcp server %q: tool discovery failed: %w", def.Alias, err)) + for _, t := range res.Tools { + if t == nil || !selected[t.Name] { + continue } - count += len(res.Tools) - if count > maxTools { - cancel() - return fail(fmt.Errorf("mcp server %q: discovery exceeds %d tools", def.Alias, maxTools)) + found[t.Name] = true + // The alias__ prefix keeps names clear of the built-in tools. + name := def.Alias + "__" + t.Name + if !validName(name) { + return fmt.Errorf("mcp server %q: external tool name %q is invalid", def.Alias, t.Name) } - for _, t := range res.Tools { - if t == nil { - continue - } - if !selected[t.Name] { - continue - } - found[t.Name] = true - name := def.Alias + "__" + t.Name - if !validName(name) || reserved[name] { - cancel() - return fail(fmt.Errorf("mcp server %q: external tool name %q is invalid or reserved", def.Alias, t.Name)) - } - if _, ok := r.tools[name]; ok { - cancel() - return fail(fmt.Errorf("mcp server %q: external tool name collision %q", def.Alias, name)) - } - params, err := schema(t.InputSchema) - if err != nil { - cancel() - return fail(fmt.Errorf("mcp server %q tool %q: unsupported input schema: %w", def.Alias, t.Name, err)) - } - encodedSchema, _ := json.Marshal(params) - schemaBytes += len(encodedSchema) - if schemaBytes > maxCatalogSchemaBytes { - cancel() - return fail(fmt.Errorf("mcp catalog exceeds %d schema bytes", maxCatalogSchemaBytes)) - } - r.tools[name] = Tool{server: sv, original: t.Name} - desc := t.Description - if desc == "" { - desc = "MCP tool from configured server " + def.Alias - } - r.Specs = append(r.Specs, model.ToolSpec{Type: "function", Function: model.FunctionSpec{Name: name, Description: desc, Parameters: params}}) - if len(r.Specs) > maxTools { - cancel() - return fail(fmt.Errorf("mcp catalog exceeds %d selected tools", maxTools)) - } + if _, ok := r.tools[name]; ok { + return fmt.Errorf("mcp server %q: external tool name collision %q", def.Alias, name) } - if res.NextCursor == "" { - break + params, err := schema(t.InputSchema) + if err != nil { + return fmt.Errorf("mcp server %q tool %q: unsupported input schema: %w", def.Alias, t.Name, err) } - if cursors[res.NextCursor] { - cancel() - return fail(fmt.Errorf("mcp server %q: tool discovery pagination loop", def.Alias)) + encodedSchema, _ := json.Marshal(params) + *schemaBytes += len(encodedSchema) + if *schemaBytes > maxCatalogSchemaBytes { + return fmt.Errorf("mcp catalog exceeds %d schema bytes", maxCatalogSchemaBytes) } - cursors[res.NextCursor] = true - cursor = res.NextCursor - if page == maxPages-1 { - cancel() - return fail(fmt.Errorf("mcp server %q: discovery exceeds %d pages", def.Alias, maxPages)) + r.tools[name] = Tool{server: sv, original: t.Name} + desc := t.Description + if desc == "" { + desc = "MCP tool from configured server " + def.Alias } - } - cancel() - for name := range selected { - if !found[name] { - return fail(fmt.Errorf("mcp server %q: selected tool %q was not discovered", def.Alias, name)) + r.Specs = append(r.Specs, model.ToolSpec{Type: "function", Function: model.FunctionSpec{Name: name, Description: desc, Parameters: params}}) + if len(r.Specs) > maxTools { + return fmt.Errorf("mcp catalog exceeds %d selected tools", maxTools) } } - fmt.Fprintf(stderr, "mcp: enabled server %s (%d selected tools)\n", def.Alias, len(selected)) + if res.NextCursor == "" { + break + } + if cursors[res.NextCursor] { + return fmt.Errorf("mcp server %q: tool discovery pagination loop", def.Alias) + } + cursors[res.NextCursor] = true + cursor = res.NextCursor } - if len(r.Specs) == 0 { - return fail(errors.New("mcp: configured servers produced no tools")) + for _, name := range def.Tools { + if !found[name] { + return fmt.Errorf("mcp server %q: selected tool %q was not discovered", def.Alias, name) + } } - return r, nil + return nil } func validateDefinitions(cfg []config.MCPServer) error { @@ -247,64 +218,69 @@ func validateDefinitions(cfg []config.MCPServer) error { } // Call executes only a catalogued tool, preserving the original JSON values. -func (r *Registry) Call(ctx context.Context, name, raw string) (string, bool) { +// ok is false when name is not catalogued; isErr marks a failed call. +func (r *Registry) Call(ctx context.Context, name, raw string) (out string, isErr, ok bool) { t, ok := r.tools[name] if !ok { - return "", false + return "", false, false } if t.server.failed { - return "error: MCP server " + t.server.alias + " connection is unavailable after an earlier failure", true + return "error: MCP server " + t.server.alias + " connection is unavailable after an earlier failure", true, true } var args map[string]any decoder := json.NewDecoder(bytes.NewBufferString(raw)) decoder.UseNumber() if err := decoder.Decode(&args); err != nil || args == nil { - return "error: invalid arguments JSON object", true + return "error: invalid arguments JSON object", true, true } if decoder.Decode(new(any)) != io.EOF { - return "error: invalid arguments JSON object", true + return "error: invalid arguments JSON object", true, true } cctx, cancel := context.WithTimeout(ctx, t.server.timeout) defer cancel() res, err := t.server.session.CallTool(cctx, &mcp.CallToolParams{Name: t.original, Arguments: args}) if err != nil { + // A timeout or interrupt cancels only this request; the session stays + // usable. Any other error means the connection itself is suspect. + if cctx.Err() != nil { + if ctx.Err() == nil { + return fmt.Sprintf("error: MCP tool %s timed out after %s", name, t.server.timeout), true, true + } + return "error: MCP tool " + name + " was interrupted: " + ctx.Err().Error(), true, true + } t.server.failed = true - return truncate("error: MCP server " + t.server.alias + " call failed: " + err.Error()), true + return truncate("error: MCP server " + t.server.alias + " call failed: " + err.Error()), true, true } - var out strings.Builder + var b strings.Builder for _, c := range res.Content { - switch v := c.(type) { - case *mcp.TextContent: - if out.Len() > 0 { - out.WriteByte('\n') - } - out.WriteString(v.Text) - default: - if out.Len() > 0 { - out.WriteByte('\n') - } - out.WriteString("[omitted unsupported MCP content]") + if b.Len() > 0 { + b.WriteByte('\n') + } + if v, text := c.(*mcp.TextContent); text { + b.WriteString(v.Text) + } else { + b.WriteString("[omitted unsupported MCP content]") } } if res.StructuredContent != nil { - b, _ := json.Marshal(res.StructuredContent) - if out.Len() > 0 { - out.WriteByte('\n') + data, _ := json.Marshal(res.StructuredContent) + if b.Len() > 0 { + b.WriteByte('\n') } - out.WriteString("structured JSON: ") - out.Write(b) + b.WriteString("structured JSON: ") + b.Write(data) } if res.NeedsInput() { - return "error: MCP tool requested unsupported interactive input", true + return "error: MCP tool requested unsupported interactive input", true, true } - text := out.String() + text := b.String() if res.IsError { text = "tool error: " + text } if text == "" { text = "[MCP tool returned no text or structured JSON]" } - return truncate(text), true + return truncate(text), res.IsError, true } func (r *Registry) Close() { @@ -360,6 +336,9 @@ func schema(v any) (map[string]any, error) { if err = json.Unmarshal(b, &m); err != nil || m == nil { return nil, errors.New("schema must be an object") } + // SDK-generated schemas commonly declare their draft; it is metadata + // only, and providers don't need it. + delete(m, "$schema") if err = checkSchema(m, 0); err != nil { return nil, err } @@ -374,23 +353,20 @@ func checkSchema(m map[string]any, depth int) error { } for k := range m { switch k { - case "type", "description", "properties", "required", "items", "additionalProperties", "enum", "title", "default", "minimum", "maximum", "minLength", "maxLength", "minItems", "maxItems": + case "type", "description", "properties", "required", "items", "additionalProperties", "enum", "const", "title", "default", "examples", + "minimum", "maximum", "exclusiveMinimum", "exclusiveMaximum", "multipleOf", + "minLength", "maxLength", "pattern", "format", "minItems", "maxItems", "uniqueItems", "minProperties", "maxProperties": default: return fmt.Errorf("keyword %q is unsupported", k) } } - t, ok := m["type"].(string) - if !ok { - return errors.New("every schema must declare a type") - } - switch t { - case "object", "array", "string", "number", "integer", "boolean", "null": - default: - return fmt.Errorf("type %q is unsupported", t) + types, err := schemaTypes(m["type"]) + if err != nil { + return err } if p, ok := m["properties"]; ok { props, yes := p.(map[string]any) - if !yes || t != "object" { + if !yes || !types["object"] { return errors.New("properties requires object type") } for _, v := range props { @@ -405,7 +381,7 @@ func checkSchema(m map[string]any, depth int) error { } if i, ok := m["items"]; ok { item, yes := i.(map[string]any) - if !yes || t != "array" { + if !yes || !types["array"] { return errors.New("items requires array type") } if err := checkSchema(item, depth+1); err != nil { @@ -413,13 +389,13 @@ func checkSchema(m map[string]any, depth int) error { } } if a, ok := m["additionalProperties"]; ok { - if _, yes := a.(bool); !yes || t != "object" { + if _, yes := a.(bool); !yes || !types["object"] { return errors.New("additionalProperties must be boolean") } } if req, ok := m["required"]; ok { arr, yes := req.([]any) - if !yes || t != "object" { + if !yes || !types["object"] { return errors.New("required must be an array") } for _, v := range arr { @@ -430,3 +406,29 @@ func checkSchema(m map[string]any, depth int) error { } return nil } + +// schemaTypes returns the set of types a schema declares: one type name, or +// an array of them such as ["string", "null"] for a nullable value. +func schemaTypes(v any) (map[string]bool, error) { + var names []any + switch t := v.(type) { + case string: + names = []any{t} + case []any: + names = t + } + if len(names) == 0 { + return nil, errors.New("every schema must declare a type") + } + types := map[string]bool{} + for _, n := range names { + name, _ := n.(string) + switch name { + case "object", "array", "string", "number", "integer", "boolean", "null": + types[name] = true + default: + return nil, fmt.Errorf("type %v is unsupported", n) + } + } + return types, nil +} diff --git a/internal/mcptools/client_test.go b/internal/mcptools/client_test.go index febdca4..7d826e7 100644 --- a/internal/mcptools/client_test.go +++ b/internal/mcptools/client_test.go @@ -4,7 +4,9 @@ import ( "bytes" "context" "os" + "strings" "testing" + "time" "github.com/modelcontextprotocol/go-sdk/mcp" "github.com/vicgupta/tinyagent/internal/config" @@ -18,30 +20,118 @@ func TestMCPFixtureProcess(t *testing.T) { mcp.AddTool[map[string]any, any](s, &mcp.Tool{Name: "echo", Description: "return arguments", InputSchema: map[string]any{"type": "object", "properties": map[string]any{"payload": map[string]any{"type": "object"}}}}, func(_ context.Context, _ *mcp.CallToolRequest, in map[string]any) (*mcp.CallToolResult, any, error) { return nil, in, nil }) + // Shaped like SDK-generated schemas: a $schema declaration, a nullable + // field, and string formats. + sdkSchema := map[string]any{ + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": map[string]any{"url": map[string]any{"type": "string", "format": "uri"}, "note": map[string]any{"type": []any{"string", "null"}}}, + "required": []any{"url"}, + "additionalProperties": false, + } + mcp.AddTool[map[string]any, any](s, &mcp.Tool{Name: "fetch", InputSchema: sdkSchema}, func(_ context.Context, _ *mcp.CallToolRequest, in map[string]any) (*mcp.CallToolResult, any, error) { + return &mcp.CallToolResult{Content: []mcp.Content{&mcp.TextContent{Text: "fetched " + in["url"].(string)}}}, nil, nil + }) + mcp.AddTool[map[string]any, any](s, &mcp.Tool{Name: "fail", InputSchema: map[string]any{"type": "object"}}, func(context.Context, *mcp.CallToolRequest, map[string]any) (*mcp.CallToolResult, any, error) { + return &mcp.CallToolResult{IsError: true, Content: []mcp.Content{&mcp.TextContent{Text: "no such page"}}}, nil, nil + }) + mcp.AddTool[map[string]any, any](s, &mcp.Tool{Name: "slow", InputSchema: map[string]any{"type": "object"}}, func(ctx context.Context, _ *mcp.CallToolRequest, _ map[string]any) (*mcp.CallToolResult, any, error) { + select { + case <-ctx.Done(): + case <-time.After(10 * time.Second): + } + return &mcp.CallToolResult{Content: []mcp.Content{&mcp.TextContent{Text: "late"}}}, nil, nil + }) + mcp.AddTool[map[string]any, any](s, &mcp.Tool{Name: "union", InputSchema: map[string]any{"type": "object", "properties": map[string]any{"v": map[string]any{"anyOf": []any{map[string]any{"type": "string"}, map[string]any{"type": "number"}}}}}}, func(context.Context, *mcp.CallToolRequest, map[string]any) (*mcp.CallToolResult, any, error) { + return nil, nil, nil + }) if err := s.Run(context.Background(), &mcp.StdioTransport{}); err != nil { os.Exit(3) } os.Exit(0) } -func TestStartAndCallPreservesJSONArguments(t *testing.T) { +// startFixture connects to the fixture server above with the given allowlist. +func startFixture(t *testing.T, def config.MCPServer) (*Registry, error) { + t.Helper() t.Setenv("TINYAGENT_MCP_FIXTURE", "1") - ctx := context.Background() + def.Alias = "docs" + def.Command = os.Args[0] + def.Args = []string{"-test.run=^TestMCPFixtureProcess$"} + def.Env = []string{"TINYAGENT_MCP_FIXTURE"} var stderr bytes.Buffer - r, err := Start(ctx, t.TempDir(), []config.MCPServer{{Alias: "docs", Command: os.Args[0], Args: []string{"-test.run=^TestMCPFixtureProcess$"}, Tools: []string{"echo"}, Env: []string{"TINYAGENT_MCP_FIXTURE"}}}, &stderr) + r, err := Start(context.Background(), t.TempDir(), []config.MCPServer{def}, &stderr) + if r != nil { + t.Cleanup(r.Close) + } + return r, err +} + +func TestStartAndCallPreservesJSONArguments(t *testing.T) { + r, err := startFixture(t, config.MCPServer{Tools: []string{"echo"}}) if err != nil { t.Fatal(err) } - defer r.Close() if len(r.Specs) != 1 || r.Specs[0].Function.Name != "docs__echo" { t.Fatalf("catalog = %#v", r.Specs) } - got, ok := r.Call(ctx, "docs__echo", `{"payload":{"nested":[true,12.5,{"key":"v"}]},"extra_name":[false]}`) - if !ok { - t.Fatal("external tool wasn't dispatched") + got, isErr, ok := r.Call(context.Background(), "docs__echo", `{"payload":{"nested":[true,12.5,{"key":"v"}]},"extra_name":[false]}`) + if !ok || isErr { + t.Fatalf("Call = %q, isErr %v, ok %v", got, isErr, ok) } const want = `{"extra_name":[false],"payload":{"nested":[true,12.5,{"key":"v"}]}}` - if !bytes.Contains([]byte(got), []byte(want)) { + if !strings.Contains(got, want) { t.Fatalf("result lost arguments: %s (want containing %s)", got, want) } + if _, _, ok := r.Call(context.Background(), "bash", `{}`); ok { + t.Error("a built-in tool name was dispatched to MCP") + } +} + +func TestSDKStyleSchemaIsAccepted(t *testing.T) { + r, err := startFixture(t, config.MCPServer{Tools: []string{"fetch"}}) + if err != nil { + t.Fatal(err) + } + params := r.Specs[0].Function.Parameters + if _, ok := params["$schema"]; ok { + t.Errorf("$schema was sent to the provider: %#v", params) + } + got, isErr, _ := r.Call(context.Background(), "docs__fetch", `{"url":"https://example.com","note":null}`) + if isErr || got != "fetched https://example.com" { + t.Errorf("Call = %q, isErr %v", got, isErr) + } +} + +func TestUnsupportedSchemaFailsStartup(t *testing.T) { + _, err := startFixture(t, config.MCPServer{Tools: []string{"union"}}) + if err == nil || !strings.Contains(err.Error(), `keyword "anyOf" is unsupported`) { + t.Fatalf("err = %v, want the anyOf keyword rejected", err) + } +} + +func TestToolErrorIsFlagged(t *testing.T) { + r, err := startFixture(t, config.MCPServer{Tools: []string{"fail"}}) + if err != nil { + t.Fatal(err) + } + got, isErr, ok := r.Call(context.Background(), "docs__fail", `{}`) + if !ok || !isErr || got != "tool error: no such page" { + t.Errorf("Call = %q, isErr %v, ok %v", got, isErr, ok) + } +} + +func TestCallTimeoutKeepsServerUsable(t *testing.T) { + r, err := startFixture(t, config.MCPServer{Tools: []string{"slow", "fetch"}, CallTimeoutSec: 1}) + if err != nil { + t.Fatal(err) + } + got, isErr, _ := r.Call(context.Background(), "docs__slow", `{}`) + if !isErr || !strings.Contains(got, "timed out after 1s") { + t.Fatalf("slow Call = %q, isErr %v", got, isErr) + } + got, isErr, _ = r.Call(context.Background(), "docs__fetch", `{"url":"u"}`) + if isErr || got != "fetched u" { + t.Errorf("call after a timeout = %q, isErr %v; want the server still usable", got, isErr) + } } diff --git a/internal/tools/tools.go b/internal/tools/tools.go index f94b459..e559e3d 100644 --- a/internal/tools/tools.go +++ b/internal/tools/tools.go @@ -36,7 +36,9 @@ type Executor struct { // stream: true. nil discards it (the result is unchanged either way). Stream io.Writer ExternalSpecs []model.ToolSpec - ExternalCall func(context.Context, string, string) (string, bool) + // ExternalCall runs a tool from ExternalSpecs. ok is false when name is + // not an external tool; isErr marks a failed call. + ExternalCall func(ctx context.Context, name, args string) (out string, isErr, ok bool) procs registry } @@ -146,8 +148,8 @@ func (e *Executor) RunWithError(ctx context.Context, call model.ToolCall) (strin // run() below then rejects them with the read-only error, matching // Catalog(), which also excludes them there. if !e.ReadOnly && e.ExternalCall != nil { - if out, ok := e.ExternalCall(ctx, call.Function.Name, call.Function.Arguments); ok { - return truncate(out, e.MaxOutputChars), externalError(out) + if out, isErr, ok := e.ExternalCall(ctx, call.Function.Name, call.Function.Arguments); ok { + return truncate(out, e.MaxOutputChars), isErr } } out, err := e.run(ctx, call.Function.Name, call.Function.Arguments) @@ -169,13 +171,6 @@ func (e *Executor) Catalog() []model.ToolSpec { return append(out, e.ExternalSpecs...) } -// externalError reports whether an external tool's output is a failure. The -// MCP client returns failures as text prefixed "error: " (call or protocol -// failures) or "tool error: " (results the server flagged with isError). -func externalError(out string) bool { - return strings.HasPrefix(out, "error: ") || strings.HasPrefix(out, "tool error: ") -} - func (e *Executor) run(ctx context.Context, name, raw string) (string, error) { if e.ReadOnly && (name != "read_file" && name != "list_files" && name != "grep") { return "", fmt.Errorf("tool %q is unavailable in read-only mode; only read_file, list_files, and grep are allowed", name)