diff --git a/agent/caller/jsapi.go b/agent/caller/jsapi.go index 2edc023c..b9077ba8 100644 --- a/agent/caller/jsapi.go +++ b/agent/caller/jsapi.go @@ -27,8 +27,12 @@ func NewJSAPI(ctx *agentContext.Context) *JSAPI { // Call executes a single agent call // Usage: ctx.agent.Call("assistant-id", messages, options?) // Returns: { agent_id, response, content, error } +// Note: SSE output is automatically disabled to prevent sub-agent from writing to +// the same SSE stream as the parent agent, which would cause client issues. func (api *JSAPI) Call(agentID string, messages []interface{}, opts map[string]interface{}) interface{} { req := api.buildRequest(agentID, messages, opts) + // Force skip.output = true for sub-agent calls + api.forceSkipOutput(req) result := api.orchestrator.callAgent(req) return result } @@ -71,9 +75,12 @@ func (api *JSAPI) Race(requests []interface{}) []interface{} { // ============================================================================ // CallWithHandler executes a single agent call with an OnMessage handler +// Note: SSE output is automatically disabled; use the handler callback to receive messages. func (api *JSAPI) CallWithHandler(agentID string, messages []interface{}, opts map[string]interface{}, handler agentContext.OnMessageFunc) interface{} { req := api.buildRequest(agentID, messages, opts) req.Handler = handler + // Force skip.output = true for sub-agent calls + api.forceSkipOutput(req) result := api.orchestrator.callAgent(req) return result } @@ -99,8 +106,24 @@ func (api *JSAPI) RaceWithHandler(requests []interface{}, globalHandler agentCon return api.convertResults(results) } +// forceSkipOutput ensures SSE output is disabled for sub-agent calls +// This prevents sub-agents from writing to the same SSE stream as the parent, +// which would cause client disconnection and message corruption. +// Users can use the onChunk callback to receive streaming messages if needed. +func (api *JSAPI) forceSkipOutput(req *Request) { + if req.Options == nil { + req.Options = &CallOptions{} + } + if req.Options.Skip == nil { + req.Options.Skip = &agentContext.Skip{} + } + req.Options.Skip.Output = true +} + // parseRequestsWithHandlers parses requests and attaches handlers // It checks for per-request _handler fields and wraps globalHandler with agentID/index +// For all calls, this automatically sets skip.output = true to prevent sub-agents +// from writing to the same SSE stream as the parent, which would cause client issues. func (api *JSAPI) parseRequestsWithHandlers(requests []interface{}, globalHandler agentContext.BatchOnMessageFunc) []*Request { reqs := make([]*Request, 0, len(requests)) @@ -130,6 +153,9 @@ func (api *JSAPI) parseRequestsWithHandlers(requests []interface{}, globalHandle req := api.buildRequest(agentID, messages, opts) + // Force skip.output = true for all sub-agent calls + api.forceSkipOutput(req) + // Check for per-request handler first (takes precedence) if handler, ok := reqMap["_handler"].(agentContext.OnMessageFunc); ok && handler != nil { req.Handler = handler diff --git a/agent/context/JSAPI.md b/agent/context/JSAPI.md index 4d6fede8..4103b244 100644 --- a/agent/context/JSAPI.md +++ b/agent/context/JSAPI.md @@ -1631,6 +1631,9 @@ The `ctx.mcp` object provides access to Model Context Protocol operations for in | `CallTool(client, name, args?)` | Call a single tool | | `CallTools(client, tools)` | Call multiple tools sequentially | | `CallToolsParallel(client, tools)` | Call multiple tools in parallel | +| `All(requests)` | Call tools across servers, wait for all | +| `Any(requests)` | Call tools across servers, first success wins | +| `Race(requests)` | Call tools across servers, first complete wins | | `ListPrompts(client, cursor?)` | List available prompts | | `GetPrompt(client, name, args?)` | Get a specific prompt | | `ListSamples(client, type, name)` | List samples for a tool/resource | @@ -1684,7 +1687,7 @@ console.log(tools.tools); // Array of tools #### `ctx.mcp.CallTool(client, name, arguments?)` -Calls a single tool. +Calls a single tool and returns the parsed result directly. **Parameters:** @@ -1692,43 +1695,61 @@ Calls a single tool. - `name`: String - Tool name - `arguments`: Object (optional) - Tool arguments +**Returns:** Parsed result directly (automatically extracts and parses JSON from tool response) + ```javascript -const result = ctx.mcp.CallTool("echo", "ping", { count: 3 }); -console.log(result.content); // Tool result content +// Result is returned directly - no wrapper object needed +const result = ctx.mcp.CallTool("echo", "echo", { message: "hello" }); +console.log(result.echo); // "hello" - directly access parsed data! + +// Another example +const status = ctx.mcp.CallTool("echo", "status", { verbose: true }); +console.log(status.status); // "online" +console.log(status.uptime); // 3600 ``` #### `ctx.mcp.CallTools(client, tools)` -Calls multiple tools sequentially. +Calls multiple tools sequentially and returns array of parsed results. **Parameters:** - `client`: String - MCP client ID - `tools`: Array - Array of tool call objects +**Returns:** Array of parsed results (same order as input tools) + ```javascript const results = ctx.mcp.CallTools("echo", [ { name: "ping", arguments: { count: 1 } }, - { name: "status", arguments: { verbose: true } }, + { name: "echo", arguments: { message: "hello" } }, ]); -console.log(results.results); // Array of results + +// Results are directly accessible +console.log(results[0].message); // "pong" +console.log(results[1].echo); // "hello" ``` #### `ctx.mcp.CallToolsParallel(client, tools)` -Calls multiple tools in parallel. +Calls multiple tools in parallel and returns array of parsed results. **Parameters:** - `client`: String - MCP client ID - `tools`: Array - Array of tool call objects +**Returns:** Array of parsed results (same order as input tools) + ```javascript const results = ctx.mcp.CallToolsParallel("echo", [ { name: "ping", arguments: { count: 1 } }, - { name: "status", arguments: { verbose: false } }, + { name: "echo", arguments: { message: "hello" } }, ]); -console.log(results.results); // Array of results (order may vary) + +// Results are directly accessible (order matches input order) +console.log(results[0].message); // "pong" (ping result) +console.log(results[1].echo); // "hello" (echo result) ``` ### Prompt Operations @@ -1797,6 +1818,127 @@ const sample = ctx.mcp.GetSample("echo", "tool", "ping", 0); console.log(sample.name, sample.input); // Sample name and input data ``` +### Cross-Server Tool Operations + +These methods enable calling tools across multiple MCP servers concurrently, similar to JavaScript Promise patterns. This is useful for: + +- **Parallel data fetching**: Query multiple data sources simultaneously +- **Redundancy/Fallback**: Try multiple servers, use first successful result +- **Load balancing**: Distribute load across servers + +#### `ctx.mcp.All(requests)` + +Calls tools on multiple MCP servers concurrently and waits for all to complete (like `Promise.all`). + +**Parameters:** + +- `requests`: Array of request objects with `mcp`, `tool`, and optional `arguments` + +**Returns:** Array of `MCPToolResult` objects in the same order as requests + +```javascript +const results = ctx.mcp.All([ + { mcp: "server1", tool: "search", arguments: { query: "topic" } }, + { mcp: "server2", tool: "fetch", arguments: { id: 123 } }, + { mcp: "server3", tool: "analyze", arguments: { data: "input" } } +]); + +// Process all results +results.forEach((r, i) => { + if (r.error) { + console.log(`Request ${i} failed: ${r.error}`); + } else { + console.log(`Request ${i} result:`, r.result); + } +}); +``` + +#### `ctx.mcp.Any(requests)` + +Calls tools on multiple MCP servers concurrently and returns when any succeeds (like `Promise.any`). Useful for redundancy/fallback scenarios. + +**Parameters:** + +- `requests`: Array of request objects + +**Returns:** Array of `MCPToolResult` objects (only contains results received before first success) + +```javascript +// Try multiple search providers, use first successful result +const results = ctx.mcp.Any([ + { mcp: "search-primary", tool: "search", arguments: { q: "query" } }, + { mcp: "search-backup", tool: "search", arguments: { q: "query" } } +]); + +const success = results.find(r => r && !r.error); +if (success) { + console.log("Search result:", success.result); +} +``` + +#### `ctx.mcp.Race(requests)` + +Calls tools on multiple MCP servers concurrently and returns when any completes (like `Promise.race`). Returns immediately with first completion, regardless of success or failure. + +**Parameters:** + +- `requests`: Array of request objects + +**Returns:** Array of `MCPToolResult` objects (only first completed result is populated) + +```javascript +// Get fastest response +const results = ctx.mcp.Race([ + { mcp: "region-us", tool: "ping", arguments: {} }, + { mcp: "region-eu", tool: "ping", arguments: {} }, + { mcp: "region-asia", tool: "ping", arguments: {} } +]); + +const first = results.find(r => r !== undefined && r !== null); +console.log(`Fastest server: ${first.mcp}`); +``` + +#### MCPToolRequest Structure + +```typescript +interface MCPToolRequest { + mcp: string; // MCP server ID (required) + tool: string; // Tool name (required) + arguments?: any; // Tool arguments (optional) +} +``` + +#### MCPToolResult Structure + +```typescript +interface MCPToolResult { + mcp: string; // MCP server ID + tool: string; // Tool name + result?: any; // Parsed result content (directly usable) + error?: string; // Error message (on failure) +} +``` + +The `result` field contains the automatically parsed content from the MCP response: +- For text content: JSON parsed if valid JSON, otherwise plain string +- For image content: `{ type: "image", data: "...", mimeType: "..." }` +- For resource content: The resource object directly +- If only one content item exists, returns it directly (not as array) + +**Example using parsed result:** + +```javascript +// Single server - direct result +const result = ctx.mcp.CallTool("echo", "echo", { message: "hello" }); +console.log(result.echo); // Directly access parsed data + +// Cross-server - results array with MCPToolResult objects +const results = ctx.mcp.All([ + { mcp: "echo", tool: "echo", arguments: { message: "hello" } } +]); +console.log(results[0].result.echo); // Access via .result field +``` + ## Agent API The `ctx.agent` object provides methods to call other agents from within hooks, enabling agent-to-agent communication (A2A). This allows building complex multi-agent workflows where agents can delegate tasks, consult specialists, or orchestrate parallel operations. @@ -1902,6 +2044,13 @@ Common message types: The parallel methods allow calling multiple agents concurrently, similar to JavaScript Promise patterns. +> **Important: SSE Output is Automatically Disabled** +> +> For all batch calls (`All`, `Any`, `Race`), SSE output is **automatically disabled** (`skip.output = true`). +> This prevents multiple agents from writing to the same SSE stream simultaneously, which would cause +> client disconnection and message corruption. Use the `onChunk` callback to receive streaming messages +> if needed. + #### `ctx.agent.All(requests, options?)` Executes all agent calls and waits for all to complete (like `Promise.all`). @@ -1922,6 +2071,7 @@ interface AgentRequest { // Note: Per-request onChunk is NOT supported in batch calls. // Use the global onChunk callback in the second argument instead. +// Note: skip.output is automatically set to true for all batch calls. ``` **Example:** diff --git a/agent/context/context.go b/agent/context/context.go index b89b4ff6..98cc3e20 100644 --- a/agent/context/context.go +++ b/agent/context/context.go @@ -133,17 +133,34 @@ func (ctx *Context) GetAuthorizedMap() map[string]interface{} { } // Fork creates a child context for concurrent agent/LLM calls -// The forked context shares read-only resources (Memory, Authorized, Cache, Writer) -// but has its own independent Stack and Logger to avoid race conditions +// The forked context shares read-only resources (Authorized, Cache, Writer) +// but has its own independent Stack, Logger, and Memory.Context namespace +// to avoid race conditions and state sharing issues. // // This is essential for batch operations (All/Any/Race) where multiple goroutines -// need to execute concurrently without interfering with each other's Stack state. +// need to execute concurrently without interfering with each other's state. +// +// Key behavior: +// - Memory.User, Memory.Team, Memory.Chat are shared (cross-request state) +// - Memory.Context is INDEPENDENT (request-scoped state, isolated per fork) // // The forked context does NOT need to be released separately - the parent context // manages shared resources. However, the child's Stack will be collected in parent's Stacks map. func (ctx *Context) Fork() *Context { childID := generateContextID() + // Fork memory with independent Context namespace + // This prevents parallel sub-agents from sharing ctx.memory.context state + var forkedMemory *memory.Memory + if ctx.Memory != nil { + var err error + forkedMemory, err = ctx.Memory.Fork(childID) + if err != nil { + // Fallback to shared memory if fork fails (log warning) + forkedMemory = ctx.Memory + } + } + child := &Context{ // Inherit parent's standard context Context: ctx.Context, @@ -151,8 +168,10 @@ func (ctx *Context) Fork() *Context { // New unique ID for this forked context ID: childID, + // Memory with independent Context namespace (see above) + Memory: forkedMemory, + // Share read-only/thread-safe resources with parent - Memory: ctx.Memory, // Memory is designed to be shared Cache: ctx.Cache, // Cache store is thread-safe Writer: ctx.Writer, // Output writer is thread-safe (output module handles concurrency) Authorized: ctx.Authorized, // Read-only auth info diff --git a/agent/context/jsapi_mcp.go b/agent/context/jsapi_mcp.go index acc82869..088f923d 100644 --- a/agent/context/jsapi_mcp.go +++ b/agent/context/jsapi_mcp.go @@ -1,11 +1,16 @@ package context import ( + "fmt" + "github.com/yaoapp/gou/mcp/types" "github.com/yaoapp/gou/runtime/v8/bridge" "rogchap.com/v8go" ) +// Suppress unused import warning - types is used in other functions +var _ = types.ToolCall{} + // MCP JavaScript API methods // These methods expose MCP functionality to JavaScript runtime @@ -101,6 +106,7 @@ func (ctx *Context) mcpListToolsMethod(iso *v8go.Isolate) *v8go.FunctionTemplate // mcpCallToolMethod implements ctx.MCP.CallTool(mcp, name, args) // Calls a specific tool from an MCP client +// Returns CallToolResult with parsed 'result' field for convenience func (ctx *Context) mcpCallToolMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { v8ctx := info.Context() @@ -125,11 +131,14 @@ func (ctx *Context) mcpCallToolMethod(iso *v8go.Isolate) *v8go.FunctionTemplate } } - result, err := ctx.CallTool(mcpID, toolName, toolArgs) + response, err := ctx.CallTool(mcpID, toolName, toolArgs) if err != nil { return bridge.JsException(v8ctx, err.Error()) } + // Return parsed result directly + result := parseCallToolResponse(response) + jsVal, err := bridge.JsValue(v8ctx, result) if err != nil { return bridge.JsException(v8ctx, err.Error()) @@ -141,6 +150,7 @@ func (ctx *Context) mcpCallToolMethod(iso *v8go.Isolate) *v8go.FunctionTemplate // mcpCallToolsMethod implements ctx.MCP.CallTools(mcp, tools) // Calls multiple tools sequentially from an MCP client +// Returns CallToolsResult with parsed 'result' field in each item func (ctx *Context) mcpCallToolsMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { v8ctx := info.Context() @@ -192,11 +202,14 @@ func (ctx *Context) mcpCallToolsMethod(iso *v8go.Isolate) *v8go.FunctionTemplate _ = i } - result, err := ctx.CallTools(mcpID, tools) + response, err := ctx.CallTools(mcpID, tools) if err != nil { return bridge.JsException(v8ctx, err.Error()) } + // Return parsed results directly as array + result := parseCallToolsResponse(response) + jsVal, err := bridge.JsValue(v8ctx, result) if err != nil { return bridge.JsException(v8ctx, err.Error()) @@ -208,6 +221,7 @@ func (ctx *Context) mcpCallToolsMethod(iso *v8go.Isolate) *v8go.FunctionTemplate // mcpCallToolsParallelMethod implements ctx.MCP.CallToolsParallel(mcp, tools) // Calls multiple tools in parallel from an MCP client +// Returns CallToolsResult with parsed 'result' field in each item func (ctx *Context) mcpCallToolsParallelMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { v8ctx := info.Context() @@ -259,11 +273,14 @@ func (ctx *Context) mcpCallToolsParallelMethod(iso *v8go.Isolate) *v8go.Function _ = i } - result, err := ctx.CallToolsParallel(mcpID, tools) + response, err := ctx.CallToolsParallel(mcpID, tools) if err != nil { return bridge.JsException(v8ctx, err.Error()) } + // Return parsed results directly as array + result := parseCallToolsResponse(response) + jsVal, err := bridge.JsValue(v8ctx, result) if err != nil { return bridge.JsException(v8ctx, err.Error()) @@ -422,6 +439,146 @@ func (ctx *Context) mcpGetSampleMethod(iso *v8go.Isolate) *v8go.FunctionTemplate }) } +// mcpAllMethod implements ctx.mcp.All(requests) +// Calls tools on multiple MCP servers concurrently and waits for all to complete (like Promise.all) +// Each request should have: { mcp: string, tool: string, arguments?: object } +func (ctx *Context) mcpAllMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { + return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { + v8ctx := info.Context() + args := info.Args() + + if len(args) < 1 { + return bridge.JsException(v8ctx, "All requires requests parameter") + } + + // Parse requests array + requests, err := ctx.parseMCPToolRequests(args[0], v8ctx) + if err != nil { + return bridge.JsException(v8ctx, err.Error()) + } + + // Execute all requests + results := ctx.CallToolAll(requests) + + // Convert results to JS value + jsVal, err := bridge.JsValue(v8ctx, results) + if err != nil { + return bridge.JsException(v8ctx, "failed to convert results: "+err.Error()) + } + + return jsVal + }) +} + +// mcpAnyMethod implements ctx.mcp.Any(requests) +// Calls tools on multiple MCP servers concurrently and returns when any succeeds (like Promise.any) +// Each request should have: { mcp: string, tool: string, arguments?: object } +func (ctx *Context) mcpAnyMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { + return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { + v8ctx := info.Context() + args := info.Args() + + if len(args) < 1 { + return bridge.JsException(v8ctx, "Any requires requests parameter") + } + + // Parse requests array + requests, err := ctx.parseMCPToolRequests(args[0], v8ctx) + if err != nil { + return bridge.JsException(v8ctx, err.Error()) + } + + // Execute requests until any succeeds + results := ctx.CallToolAny(requests) + + // Convert results to JS value + jsVal, err := bridge.JsValue(v8ctx, results) + if err != nil { + return bridge.JsException(v8ctx, "failed to convert results: "+err.Error()) + } + + return jsVal + }) +} + +// mcpRaceMethod implements ctx.mcp.Race(requests) +// Calls tools on multiple MCP servers concurrently and returns when any completes (like Promise.race) +// Each request should have: { mcp: string, tool: string, arguments?: object } +func (ctx *Context) mcpRaceMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { + return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { + v8ctx := info.Context() + args := info.Args() + + if len(args) < 1 { + return bridge.JsException(v8ctx, "Race requires requests parameter") + } + + // Parse requests array + requests, err := ctx.parseMCPToolRequests(args[0], v8ctx) + if err != nil { + return bridge.JsException(v8ctx, err.Error()) + } + + // Execute requests and return first completion + results := ctx.CallToolRace(requests) + + // Convert results to JS value + jsVal, err := bridge.JsValue(v8ctx, results) + if err != nil { + return bridge.JsException(v8ctx, "failed to convert results: "+err.Error()) + } + + return jsVal + }) +} + +// parseMCPToolRequests parses JS requests array into MCPToolRequest slice +func (ctx *Context) parseMCPToolRequests(arg *v8go.Value, v8ctx *v8go.Context) ([]*MCPToolRequest, error) { + goVal, err := bridge.GoValue(arg, v8ctx) + if err != nil { + return nil, fmt.Errorf("invalid requests: %w", err) + } + + requestsArray, ok := goVal.([]interface{}) + if !ok { + return nil, fmt.Errorf("requests must be an array") + } + + requests := make([]*MCPToolRequest, 0, len(requestsArray)) + for _, item := range requestsArray { + reqMap, ok := item.(map[string]interface{}) + if !ok { + return nil, fmt.Errorf("each request must be an object") + } + + // Required: mcp + mcpID, ok := reqMap["mcp"].(string) + if !ok || mcpID == "" { + return nil, fmt.Errorf("request.mcp is required and must be a string") + } + + // Required: tool + tool, ok := reqMap["tool"].(string) + if !ok || tool == "" { + return nil, fmt.Errorf("request.tool is required and must be a string") + } + + req := &MCPToolRequest{ + MCP: mcpID, + Tool: tool, + } + + // Optional: arguments + if args, exists := reqMap["arguments"]; exists && args != nil { + req.Arguments = args + } + + requests = append(requests, req) + } + + return requests, nil +} + // newMCPObject creates a new MCP object with all MCP methods func (ctx *Context) newMCPObject(iso *v8go.Isolate) *v8go.ObjectTemplate { mcpObj := v8go.NewObjectTemplate(iso) @@ -436,6 +593,11 @@ func (ctx *Context) newMCPObject(iso *v8go.Isolate) *v8go.ObjectTemplate { mcpObj.Set("CallTools", ctx.mcpCallToolsMethod(iso)) mcpObj.Set("CallToolsParallel", ctx.mcpCallToolsParallelMethod(iso)) + // Cross-server parallel tool operations (Promise-like patterns) + mcpObj.Set("All", ctx.mcpAllMethod(iso)) + mcpObj.Set("Any", ctx.mcpAnyMethod(iso)) + mcpObj.Set("Race", ctx.mcpRaceMethod(iso)) + // Prompt operations mcpObj.Set("ListPrompts", ctx.mcpListPromptsMethod(iso)) mcpObj.Set("GetPrompt", ctx.mcpGetPromptMethod(iso)) diff --git a/agent/context/jsapi_mcp_test.go b/agent/context/jsapi_mcp_test.go index 49dd5972..803311f5 100644 --- a/agent/context/jsapi_mcp_test.go +++ b/agent/context/jsapi_mcp_test.go @@ -142,16 +142,16 @@ func TestMCPCallTool(t *testing.T) { res, err := v8.Call(v8.CallOptions{}, ` function test(ctx) { - // Call ping tool + // Call ping tool - returns parsed result directly const result = ctx.mcp.CallTool("echo", "ping", { count: 3, message: "test" }) - if (!result || !result.content) { - throw new Error("Expected content") + if (result === undefined || result === null) { + throw new Error("Expected result") } return { - has_content: result.content.length > 0, - is_error: result.isError || false + has_result: true, + message: result.message } }`, ctx) @@ -164,8 +164,8 @@ func TestMCPCallTool(t *testing.T) { t.Fatalf("Expected map result, got %T", res) } - assert.Equal(t, true, result["has_content"], "should have content") - assert.Equal(t, false, result["is_error"], "should not be error") + assert.Equal(t, true, result["has_result"], "should have result") + assert.Equal(t, "test", result["message"], "should have message") } // TestMCPCallTools tests MCP.CallTools from JavaScript @@ -177,21 +177,22 @@ func TestMCPCallTools(t *testing.T) { res, err := v8.Call(v8.CallOptions{}, ` function test(ctx) { - // Call multiple tools sequentially + // Call multiple tools sequentially - returns array of parsed results const tools = [ { name: "ping", arguments: { count: 1 } }, { name: "status", arguments: { verbose: false } } ] - const result = ctx.mcp.CallTools("echo", tools) + const results = ctx.mcp.CallTools("echo", tools) - if (!result || !result.results) { - throw new Error("Expected results") + if (!Array.isArray(results)) { + throw new Error("Expected array of results") } return { - count: result.results.length, - all_success: result.results.every(r => !r.isError) + count: results.length, + ping_message: results[0]?.message, + status_online: results[1]?.status === "online" } }`, ctx) @@ -205,7 +206,8 @@ func TestMCPCallTools(t *testing.T) { } assert.Equal(t, float64(2), result["count"], "should have 2 results") - assert.Equal(t, true, result["all_success"], "all calls should succeed") + assert.Equal(t, "pong", result["ping_message"], "ping should return pong") + assert.Equal(t, true, result["status_online"], "status should be online") } // TestMCPCallToolsParallel tests MCP.CallToolsParallel from JavaScript @@ -217,21 +219,22 @@ func TestMCPCallToolsParallel(t *testing.T) { res, err := v8.Call(v8.CallOptions{}, ` function test(ctx) { - // Call multiple tools in parallel + // Call multiple tools in parallel - returns array of parsed results const tools = [ { name: "ping", arguments: { count: 1 } }, { name: "status", arguments: { verbose: true } } ] - const result = ctx.mcp.CallToolsParallel("echo", tools) + const results = ctx.mcp.CallToolsParallel("echo", tools) - if (!result || !result.results) { - throw new Error("Expected results") + if (!Array.isArray(results)) { + throw new Error("Expected array of results") } return { - count: result.results.length, - all_success: result.results.every(r => !r.isError) + count: results.length, + ping_message: results[0]?.message, + status_online: results[1]?.status === "online" } }`, ctx) @@ -245,7 +248,8 @@ func TestMCPCallToolsParallel(t *testing.T) { } assert.Equal(t, float64(2), result["count"], "should have 2 results") - assert.Equal(t, true, result["all_success"], "all calls should succeed") + assert.Equal(t, "pong", result["ping_message"], "ping should return pong") + assert.Equal(t, true, result["status_online"], "status should be online") } // TestMCPListPrompts tests MCP.ListPrompts from JavaScript @@ -404,14 +408,14 @@ func TestMCPJsApiWithTrace(t *testing.T) { // Get trace (property, not method call) const trace = ctx.trace - // Call MCP tool - should create trace node + // Call MCP tool - returns parsed result directly const result = ctx.mcp.CallTool("echo", "ping", { count: 5 }) // Verify trace and result exist return { has_trace: !!trace, - has_result: !!result, - has_content: result.content && result.content.length > 0 + has_result: result !== undefined && result !== null, + ping_message: result?.message } }`, ctx) @@ -426,5 +430,5 @@ func TestMCPJsApiWithTrace(t *testing.T) { assert.Equal(t, true, result["has_trace"], "should have trace") assert.Equal(t, true, result["has_result"], "should have result") - assert.Equal(t, true, result["has_content"], "should have content") + assert.Equal(t, "pong", result["ping_message"], "should have ping response") } diff --git a/agent/context/jsapi_mcp_v8_test.go b/agent/context/jsapi_mcp_v8_test.go new file mode 100644 index 00000000..1b72e995 --- /dev/null +++ b/agent/context/jsapi_mcp_v8_test.go @@ -0,0 +1,625 @@ +package context_test + +import ( + stdContext "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + v8 "github.com/yaoapp/gou/runtime/v8" + "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/testutils" + "github.com/yaoapp/yao/openapi/oauth/types" +) + +// TestMCP_All_V8 tests ctx.mcp.All() with real V8 execution +func TestMCP_All_V8(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + + authorized := &types.AuthorizedInfo{ + Subject: "test-user", + UserID: "test-123", + TenantID: "test-tenant", + } + + ctx := context.New(stdContext.Background(), authorized, "test-chat-mcp-all") + ctx.AssistantID = "tests.agent-caller" + defer ctx.Release() + + res, err := v8.Call(v8.CallOptions{}, ` + function test(ctx) { + try { + const results = ctx.mcp.All([ + { mcp: "echo", tool: "ping", arguments: { count: 1 } }, + { mcp: "echo", tool: "status", arguments: { verbose: false } }, + { mcp: "echo", tool: "echo", arguments: { message: "hello" } } + ]); + + return { + success: true, + count: results.length, + // Each result has mcp, tool, result (parsed), error + results: results.map(r => ({ + mcp: r.mcp, + tool: r.tool, + has_result: r.result !== undefined && r.result !== null, + error: r.error || "" + })) + }; + } catch (error) { + return { success: false, error: error.message }; + } + }`, ctx) + + require.NoError(t, err) + result, ok := res.(map[string]interface{}) + require.True(t, ok, "Result should be a map") + + success, _ := result["success"].(bool) + if !success { + t.Fatalf("Test failed: %v", result["error"]) + } + + // Should have 3 results - handle different integer types + var count int + switch v := result["count"].(type) { + case int: + count = v + case int32: + count = int(v) + case int64: + count = int(v) + case float64: + count = int(v) + default: + t.Logf("Unexpected count type: %T, value: %v", result["count"], result["count"]) + } + assert.Equal(t, 3, count, "Should have 3 results") + + // Check each result + results, ok := result["results"].([]interface{}) + require.True(t, ok, "Results should be an array") + require.Len(t, results, 3) + + for i, r := range results { + resMap, ok := r.(map[string]interface{}) + require.True(t, ok, "Result %d should be a map", i) + + hasResult, _ := resMap["has_result"].(bool) + assert.True(t, hasResult, "Result %d should have parsed result", i) + + errorStr, _ := resMap["error"].(string) + assert.Empty(t, errorStr, "Result %d should not have error", i) + } +} + +// TestMCP_All_WithError_V8 tests ctx.mcp.All() with some failing requests +func TestMCP_All_WithError_V8(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + + authorized := &types.AuthorizedInfo{ + Subject: "test-user", + UserID: "test-123", + TenantID: "test-tenant", + } + + ctx := context.New(stdContext.Background(), authorized, "test-chat-mcp-all-error") + ctx.AssistantID = "tests.agent-caller" + defer ctx.Release() + + res, err := v8.Call(v8.CallOptions{}, ` + function test(ctx) { + try { + const results = ctx.mcp.All([ + { mcp: "echo", tool: "ping", arguments: { count: 1 } }, + { mcp: "nonexistent-mcp", tool: "some-tool", arguments: {} }, + { mcp: "echo", tool: "status", arguments: {} } + ]); + + return { + success: true, + count: results.length, + results: results.map(r => ({ + mcp: r.mcp, + tool: r.tool, + has_result: r.result !== undefined && r.result !== null, + has_error: r.error !== undefined && r.error !== "" && r.error !== null + })) + }; + } catch (error) { + return { success: false, error: error.message }; + } + }`, ctx) + + require.NoError(t, err) + result, ok := res.(map[string]interface{}) + require.True(t, ok, "Result should be a map") + + success, _ := result["success"].(bool) + if !success { + t.Fatalf("Test failed: %v", result["error"]) + } + + // Should have 3 results - handle different integer types + var count int + switch v := result["count"].(type) { + case int: + count = v + case int32: + count = int(v) + case int64: + count = int(v) + case float64: + count = int(v) + } + assert.Equal(t, 3, count, "Should have 3 results") + + // Check results + results, ok := result["results"].([]interface{}) + require.True(t, ok, "Results should be an array") + + // First result (ping) should succeed + r0, _ := results[0].(map[string]interface{}) + assert.True(t, r0["has_result"].(bool), "Ping should have result") + assert.False(t, r0["has_error"].(bool), "Ping should not have error") + + // Second result (nonexistent) should fail + r1, _ := results[1].(map[string]interface{}) + assert.False(t, r1["has_result"].(bool), "Nonexistent should not have result") + assert.True(t, r1["has_error"].(bool), "Nonexistent should have error") + + // Third result (status) should succeed + r2, _ := results[2].(map[string]interface{}) + assert.True(t, r2["has_result"].(bool), "Status should have result") + assert.False(t, r2["has_error"].(bool), "Status should not have error") +} + +// TestMCP_Any_V8 tests ctx.mcp.Any() with real V8 execution +func TestMCP_Any_V8(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + + authorized := &types.AuthorizedInfo{ + Subject: "test-user", + UserID: "test-123", + TenantID: "test-tenant", + } + + ctx := context.New(stdContext.Background(), authorized, "test-chat-mcp-any") + ctx.AssistantID = "tests.agent-caller" + defer ctx.Release() + + res, err := v8.Call(v8.CallOptions{}, ` + function test(ctx) { + try { + const results = ctx.mcp.Any([ + { mcp: "echo", tool: "ping", arguments: { count: 1 } }, + { mcp: "echo", tool: "status", arguments: {} } + ]); + + // Find success results (has result field, no error) + const successResults = results.filter(r => r && r.result && !r.error); + + return { + success: true, + total_count: results.length, + success_count: successResults.length, + has_at_least_one_success: successResults.length >= 1 + }; + } catch (error) { + return { success: false, error: error.message }; + } + }`, ctx) + + require.NoError(t, err) + result, ok := res.(map[string]interface{}) + require.True(t, ok, "Result should be a map") + + success, _ := result["success"].(bool) + if !success { + t.Fatalf("Test failed: %v", result["error"]) + } + + hasAtLeastOne, _ := result["has_at_least_one_success"].(bool) + assert.True(t, hasAtLeastOne, "Should have at least one successful result") +} + +// TestMCP_Any_AllFail_V8 tests ctx.mcp.Any() when all requests fail +func TestMCP_Any_AllFail_V8(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + + authorized := &types.AuthorizedInfo{ + Subject: "test-user", + UserID: "test-123", + TenantID: "test-tenant", + } + + ctx := context.New(stdContext.Background(), authorized, "test-chat-mcp-any-fail") + ctx.AssistantID = "tests.agent-caller" + defer ctx.Release() + + res, err := v8.Call(v8.CallOptions{}, ` + function test(ctx) { + try { + const results = ctx.mcp.Any([ + { mcp: "nonexistent-1", tool: "tool1", arguments: {} }, + { mcp: "nonexistent-2", tool: "tool2", arguments: {} } + ]); + + // All should fail + const failedResults = results.filter(r => r && r.error); + + return { + success: true, + total_count: results.length, + failed_count: failedResults.length, + all_failed: failedResults.length === results.length + }; + } catch (error) { + return { success: false, error: error.message }; + } + }`, ctx) + + require.NoError(t, err) + result, ok := res.(map[string]interface{}) + require.True(t, ok, "Result should be a map") + + success, _ := result["success"].(bool) + if !success { + t.Fatalf("Test failed: %v", result["error"]) + } + + allFailed, _ := result["all_failed"].(bool) + assert.True(t, allFailed, "All requests should fail") +} + +// TestMCP_Race_V8 tests ctx.mcp.Race() with real V8 execution +func TestMCP_Race_V8(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + + authorized := &types.AuthorizedInfo{ + Subject: "test-user", + UserID: "test-123", + TenantID: "test-tenant", + } + + ctx := context.New(stdContext.Background(), authorized, "test-chat-mcp-race") + ctx.AssistantID = "tests.agent-caller" + defer ctx.Release() + + res, err := v8.Call(v8.CallOptions{}, ` + function test(ctx) { + try { + const results = ctx.mcp.Race([ + { mcp: "echo", tool: "ping", arguments: { count: 1 } }, + { mcp: "echo", tool: "status", arguments: {} } + ]); + + // Find completed results (could be success or error) + const completedResults = results.filter(r => r !== undefined && r !== null); + + return { + success: true, + total_count: results.length, + completed_count: completedResults.length, + has_at_least_one_completed: completedResults.length >= 1 + }; + } catch (error) { + return { success: false, error: error.message }; + } + }`, ctx) + + require.NoError(t, err) + result, ok := res.(map[string]interface{}) + require.True(t, ok, "Result should be a map") + + success, _ := result["success"].(bool) + if !success { + t.Fatalf("Test failed: %v", result["error"]) + } + + hasAtLeastOne, _ := result["has_at_least_one_completed"].(bool) + assert.True(t, hasAtLeastOne, "Should have at least one completed result") +} + +// TestMCP_All_ResultContent_V8 tests that the result contains parsed content directly +func TestMCP_All_ResultContent_V8(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + + authorized := &types.AuthorizedInfo{ + Subject: "test-user", + UserID: "test-123", + TenantID: "test-tenant", + } + + ctx := context.New(stdContext.Background(), authorized, "test-chat-mcp-content") + ctx.AssistantID = "tests.agent-caller" + defer ctx.Release() + + res, err := v8.Call(v8.CallOptions{}, ` + function test(ctx) { + try { + const results = ctx.mcp.All([ + { mcp: "echo", tool: "echo", arguments: { message: "hello world", uppercase: true } } + ]); + + if (results.length !== 1) { + return { success: false, error: "Expected 1 result" }; + } + + const r = results[0]; + if (r.error) { + return { success: false, error: "Tool call failed: " + r.error }; + } + + // Result should contain parsed data directly + const data = r.result; + if (!data) { + return { success: false, error: "Result should have parsed data" }; + } + + return { + success: true, + echo_message: data.echo, + uppercase_flag: data.uppercase, + original_length: data.length + }; + } catch (error) { + return { success: false, error: error.message }; + } + }`, ctx) + + require.NoError(t, err) + result, ok := res.(map[string]interface{}) + require.True(t, ok, "Result should be a map") + + success, _ := result["success"].(bool) + if !success { + t.Fatalf("Test failed: %v", result["error"]) + } + + echoMessage, _ := result["echo_message"].(string) + assert.Equal(t, "HELLO WORLD", echoMessage, "Echo message should be uppercase") + + uppercaseFlag, _ := result["uppercase_flag"].(bool) + assert.True(t, uppercaseFlag, "Uppercase flag should be true") + + // Handle different integer types from V8 + var originalLength int + switch v := result["original_length"].(type) { + case int: + originalLength = v + case int32: + originalLength = int(v) + case int64: + originalLength = int(v) + case float64: + originalLength = int(v) + } + assert.Equal(t, 11, originalLength, "Original message length should be 11") +} + +// TestMCP_All_MultipleTools_V8 tests All with multiple tools and verifies parsed results +func TestMCP_All_MultipleTools_V8(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + + authorized := &types.AuthorizedInfo{ + Subject: "test-user", + UserID: "test-123", + TenantID: "test-tenant", + } + + ctx := context.New(stdContext.Background(), authorized, "test-chat-mcp-multi") + ctx.AssistantID = "tests.agent-caller" + defer ctx.Release() + + res, err := v8.Call(v8.CallOptions{}, ` + function test(ctx) { + try { + const results = ctx.mcp.All([ + { mcp: "echo", tool: "ping", arguments: { count: 5 } }, + { mcp: "echo", tool: "echo", arguments: { message: "test", uppercase: false } } + ]); + + if (results.length !== 2) { + return { success: false, error: "Expected 2 results" }; + } + + // Access parsed results directly + const ping = results[0]; + const echo = results[1]; + + return { + success: true, + ping_message: ping.result?.message, + echo_message: echo.result?.echo + }; + } catch (error) { + return { success: false, error: error.message }; + } + }`, ctx) + + require.NoError(t, err) + result, ok := res.(map[string]interface{}) + require.True(t, ok, "Result should be a map") + + success, _ := result["success"].(bool) + if !success { + t.Fatalf("Test failed: %v", result["error"]) + } + + pingMessage, _ := result["ping_message"].(string) + assert.Equal(t, "pong", pingMessage, "Ping should return pong") + + echoMessage, _ := result["echo_message"].(string) + assert.Equal(t, "test", echoMessage, "Echo should return the message") +} + +// TestMCP_CallTool_ParsedResult_V8 tests that ctx.mcp.CallTool() returns parsed result directly +func TestMCP_CallTool_ParsedResult_V8(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + + authorized := &types.AuthorizedInfo{ + Subject: "test-user", + UserID: "test-123", + TenantID: "test-tenant", + } + + ctx := context.New(stdContext.Background(), authorized, "test-chat-mcp-calltool-parsed") + ctx.AssistantID = "tests.agent-caller" + defer ctx.Release() + + res, err := v8.Call(v8.CallOptions{}, ` + function test(ctx) { + try { + // Test CallTool returns parsed result directly + const result = ctx.mcp.CallTool("echo", "echo", { + message: "test message", + uppercase: true + }); + + // Result should be the parsed data directly + if (result === undefined || result === null) { + return { success: false, error: "Result should not be null" }; + } + + return { + success: true, + echo_message: result.echo, + original_length: result.length + }; + } catch (error) { + return { success: false, error: error.message }; + } + }`, ctx) + + require.NoError(t, err) + result, ok := res.(map[string]interface{}) + require.True(t, ok, "Result should be a map") + + success, _ := result["success"].(bool) + if !success { + t.Fatalf("Test failed: %v", result["error"]) + } + + echoMessage, _ := result["echo_message"].(string) + assert.Equal(t, "TEST MESSAGE", echoMessage, "Echo message should be uppercase") + + // Handle different integer types from V8 + var length int + switch v := result["original_length"].(type) { + case int: + length = v + case int32: + length = int(v) + case int64: + length = int(v) + case float64: + length = int(v) + } + assert.Equal(t, 12, length, "Original message length should be 12") +} + +// TestMCP_CallToolsParallel_ParsedResult_V8 tests that ctx.mcp.CallToolsParallel() returns parsed results directly +func TestMCP_CallToolsParallel_ParsedResult_V8(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + + authorized := &types.AuthorizedInfo{ + Subject: "test-user", + UserID: "test-123", + TenantID: "test-tenant", + } + + ctx := context.New(stdContext.Background(), authorized, "test-chat-mcp-calltools-parsed") + ctx.AssistantID = "tests.agent-caller" + defer ctx.Release() + + res, err := v8.Call(v8.CallOptions{}, ` + function test(ctx) { + try { + // Test CallToolsParallel returns parsed results directly as array + const results = ctx.mcp.CallToolsParallel("echo", [ + { name: "ping", arguments: { count: 2 } }, + { name: "echo", arguments: { message: "hello", uppercase: false } } + ]); + + if (!Array.isArray(results)) { + return { success: false, error: "Results should be an array" }; + } + if (results.length !== 2) { + return { success: false, error: "Expected 2 results, got " + results.length }; + } + + // Each result is the parsed data directly + const pingResult = results[0]; + const echoResult = results[1]; + + return { + success: true, + ping_message: pingResult?.message, + echo_message: echoResult?.echo + }; + } catch (error) { + return { success: false, error: error.message }; + } + }`, ctx) + + require.NoError(t, err) + result, ok := res.(map[string]interface{}) + require.True(t, ok, "Result should be a map") + + success, _ := result["success"].(bool) + if !success { + t.Fatalf("Test failed: %v", result["error"]) + } + + pingMessage, _ := result["ping_message"].(string) + assert.Equal(t, "pong", pingMessage, "Ping result should have message='pong'") + + echoMessage, _ := result["echo_message"].(string) + assert.Equal(t, "hello", echoMessage, "Echo message should be preserved (no uppercase)") +} diff --git a/agent/context/mcp.go b/agent/context/mcp.go index 254fb742..fb81c59f 100644 --- a/agent/context/mcp.go +++ b/agent/context/mcp.go @@ -3,6 +3,7 @@ package context import ( "fmt" + jsoniter "github.com/json-iterator/go" "github.com/yaoapp/gou/mcp" "github.com/yaoapp/gou/mcp/types" "github.com/yaoapp/yao/agent/i18n" @@ -570,3 +571,264 @@ func (ctx *Context) GetSample(mcpID string, itemType types.SampleItemType, itemN return result, nil } + +// Single-Server Tool Response Helpers +// ==================================== + +// parseCallToolResponse parses a CallToolResponse and returns the parsed content directly +func parseCallToolResponse(response *types.CallToolResponse) interface{} { + if response == nil { + return nil + } + return parseToolResponseContent(response) +} + +// parseCallToolsResponse parses a CallToolsResponse and returns an array of parsed results +func parseCallToolsResponse(response *types.CallToolsResponse) []interface{} { + if response == nil { + return nil + } + results := make([]interface{}, len(response.Results)) + for i, r := range response.Results { + results[i] = parseToolResponseContent(&r) + } + return results +} + +// Cross-Server Tool Operations +// ============================ + +// MCPToolRequest represents a request to call a tool on a specific MCP server +type MCPToolRequest struct { + MCP string `json:"mcp"` // MCP server ID + Tool string `json:"tool"` // Tool name + Arguments interface{} `json:"arguments"` // Tool arguments +} + +// MCPToolResult represents the result of a cross-server tool call +// Returns parsed result directly, with error field for failures +type MCPToolResult struct { + MCP string `json:"mcp"` // MCP server ID + Tool string `json:"tool"` // Tool name + Result interface{} `json:"result,omitempty"` // Parsed result content (directly usable) + Error string `json:"error,omitempty"` // Error message (on failure) +} + +// callToolResult is used internally to pass results through channels +type callToolResult struct { + idx int + result *MCPToolResult +} + +// CallToolAll calls tools on multiple MCP servers concurrently and waits for all to complete +// Returns results in the same order as requests, regardless of completion order (like Promise.all) +func (ctx *Context) CallToolAll(requests []*MCPToolRequest) []*MCPToolResult { + if len(requests) == 0 { + return []*MCPToolResult{} + } + + results := make([]*MCPToolResult, len(requests)) + done := make(chan struct{}) + remaining := len(requests) + + for i, req := range requests { + go func(idx int, r *MCPToolRequest) { + defer func() { + if err := recover(); err != nil { + results[idx] = &MCPToolResult{ + MCP: r.MCP, + Tool: r.Tool, + Error: fmt.Sprintf("panic: %v", err), + } + } + done <- struct{}{} + }() + + results[idx] = ctx.callToolSingle(r) + }(i, req) + } + + // Wait for all to complete + for remaining > 0 { + <-done + remaining-- + } + + return results +} + +// CallToolAny calls tools on multiple MCP servers concurrently and returns when any succeeds +// Returns all results received so far when first success is found (like Promise.any) +func (ctx *Context) CallToolAny(requests []*MCPToolRequest) []*MCPToolResult { + if len(requests) == 0 { + return []*MCPToolResult{} + } + + resultChan := make(chan callToolResult, len(requests)) + remaining := len(requests) + + for i, req := range requests { + go func(idx int, r *MCPToolRequest) { + defer func() { + if err := recover(); err != nil { + resultChan <- callToolResult{ + idx: idx, + result: &MCPToolResult{ + MCP: r.MCP, + Tool: r.Tool, + Error: fmt.Sprintf("panic: %v", err), + }, + } + } + }() + + resultChan <- callToolResult{idx: idx, result: ctx.callToolSingle(r)} + }(i, req) + } + + // Collect results until we find a success or all fail + results := make([]*MCPToolResult, len(requests)) + + for remaining > 0 { + cr := <-resultChan + remaining-- + results[cr.idx] = cr.result + + // Check if this is a success (no error) + if cr.result.Error == "" { + break // Stop waiting, we have a success + } + } + + // Drain remaining results in background (don't block) + if remaining > 0 { + go func(count int) { + for i := 0; i < count; i++ { + <-resultChan + } + }(remaining) + } + + return results +} + +// CallToolRace calls tools on multiple MCP servers concurrently and returns when any completes +// Returns all results received so far when first completion (like Promise.race) +func (ctx *Context) CallToolRace(requests []*MCPToolRequest) []*MCPToolResult { + if len(requests) == 0 { + return []*MCPToolResult{} + } + + resultChan := make(chan callToolResult, len(requests)) + remaining := len(requests) + + for i, req := range requests { + go func(idx int, r *MCPToolRequest) { + defer func() { + if err := recover(); err != nil { + resultChan <- callToolResult{ + idx: idx, + result: &MCPToolResult{ + MCP: r.MCP, + Tool: r.Tool, + Error: fmt.Sprintf("panic: %v", err), + }, + } + } + }() + + resultChan <- callToolResult{idx: idx, result: ctx.callToolSingle(r)} + }(i, req) + } + + // Get first result (success or failure) + results := make([]*MCPToolResult, len(requests)) + cr := <-resultChan + remaining-- + results[cr.idx] = cr.result + + // Drain remaining results in background (don't block) + if remaining > 0 { + go func(count int) { + for i := 0; i < count; i++ { + <-resultChan + } + }(remaining) + } + + return results +} + +// callToolSingle executes a single tool call on an MCP server +// This is a helper method for the parallel call methods +func (ctx *Context) callToolSingle(req *MCPToolRequest) *MCPToolResult { + result := &MCPToolResult{ + MCP: req.MCP, + Tool: req.Tool, + } + + // Call the tool using existing CallTool method + response, err := ctx.CallTool(req.MCP, req.Tool, req.Arguments) + if err != nil { + result.Error = err.Error() + return result + } + + // Parse and return result directly + result.Result = parseToolResponseContent(response) + return result +} + +// parseToolResponseContent extracts and parses the actual content from a CallToolResponse +// Similar to ToolCallResult.ParsedContent() in assistant/types.go +// - For "text" type, parses the Text field as JSON (or returns as string if not JSON) +// - For "image" type, returns the Data and MimeType +// - For "resource" type, returns the Resource object +// - If only one content item, returns it directly (not as array) +func parseToolResponseContent(response *types.CallToolResponse) interface{} { + if response == nil || len(response.Content) == 0 { + return nil + } + + var results []interface{} + for _, tc := range response.Content { + switch tc.Type { + case types.ToolContentTypeText: + // For text type, try to parse as JSON + if tc.Text != "" { + var parsed interface{} + if err := jsoniter.UnmarshalFromString(tc.Text, &parsed); err == nil { + results = append(results, parsed) + } else { + // If not JSON, return as plain string + results = append(results, tc.Text) + } + } + case types.ToolContentTypeImage: + // For image type, return data and mimeType + results = append(results, map[string]interface{}{ + "type": "image", + "data": tc.Data, + "mimeType": tc.MimeType, + }) + case types.ToolContentTypeResource: + // For resource type, return the resource object + if tc.Resource != nil { + results = append(results, tc.Resource) + } + default: + // Unknown type, include as-is with type info + results = append(results, map[string]interface{}{ + "type": tc.Type, + "text": tc.Text, + }) + } + } + + // If only one result, return it directly (not as array) + if len(results) == 1 { + return results[0] + } + + return results +} diff --git a/agent/docs/context-api.md b/agent/docs/context-api.md index 97fa70e0..b3a9a49e 100644 --- a/agent/docs/context-api.md +++ b/agent/docs/context-api.md @@ -218,20 +218,53 @@ const child = parent.Add({}, { label: "Child" }); // List tools const tools = ctx.mcp.ListTools("server-id"); -// Call single tool +// Call single tool - returns parsed result directly const result = ctx.mcp.CallTool("server-id", "tool-name", { arg: "value" }); +console.log(result.field); // Direct access to parsed data -// Call multiple sequentially +// Call multiple sequentially - returns array of parsed results const results = ctx.mcp.CallTools("server-id", [ { name: "tool1", arguments: { a: 1 } }, { name: "tool2", arguments: { b: 2 } } ]); +results.forEach(r => console.log(r)); -// Call multiple in parallel +// Call multiple in parallel - returns array of parsed results const results = ctx.mcp.CallToolsParallel("server-id", [ { name: "tool1", arguments: {} }, { name: "tool2", arguments: {} } ]); +results.forEach(r => console.log(r)); +``` + +### Cross-Server Tool Calls + +```typescript +// Call tools across multiple MCP servers (like Promise.all) +const results = ctx.mcp.All([ + { mcp: "server1", tool: "search", arguments: { q: "query" } }, + { mcp: "server2", tool: "fetch", arguments: { id: 123 } } +]); + +// First success wins (like Promise.any) +const results = ctx.mcp.Any([ + { mcp: "primary", tool: "search", arguments: { q: "query" } }, + { mcp: "backup", tool: "search", arguments: { q: "query" } } +]); + +// First complete wins (like Promise.race) +const results = ctx.mcp.Race([ + { mcp: "region-us", tool: "ping", arguments: {} }, + { mcp: "region-eu", tool: "ping", arguments: {} } +]); + +// Result structure +interface MCPToolResult { + mcp: string; // Server ID + tool: string; // Tool name + result?: any; // Parsed result content + error?: string; // Error if failed +} ``` ### Resources diff --git a/agent/docs/mcp.md b/agent/docs/mcp.md index af33e977..f05cde94 100644 --- a/agent/docs/mcp.md +++ b/agent/docs/mcp.md @@ -152,27 +152,30 @@ const tools = ctx.mcp.ListTools("server-id"); ### Call Tool ```typescript +// Returns parsed result directly - no wrapper object const result = ctx.mcp.CallTool("server-id", "search", { query: "example", limit: 10, }); -// { content: [{ type: "text", text: "..." }] } +console.log(result.items); // Direct access to parsed data ``` ### Batch Tool Calls ```typescript -// Sequential +// Sequential - returns array of parsed results const results = ctx.mcp.CallTools("server-id", [ { name: "step1", arguments: { input: "a" } }, { name: "step2", arguments: { input: "b" } }, ]); +results.forEach(r => console.log(r)); -// Parallel +// Parallel - returns array of parsed results const results = ctx.mcp.CallToolsParallel("server-id", [ { name: "api1", arguments: {} }, { name: "api2", arguments: {} }, ]); +results.forEach(r => console.log(r)); ``` ### Read Resources @@ -189,6 +192,39 @@ const prompts = ctx.mcp.ListPrompts("server-id"); const prompt = ctx.mcp.GetPrompt("server-id", "system", { role: "helper" }); ``` +### Cross-Server Tool Calls + +Call tools across multiple MCP servers concurrently: + +```typescript +// Wait for all (like Promise.all) +const results = ctx.mcp.All([ + { mcp: "server1", tool: "search", arguments: { q: "query" } }, + { mcp: "server2", tool: "analyze", arguments: { data: "input" } } +]); + +// First success (like Promise.any) - good for fallback +const results = ctx.mcp.Any([ + { mcp: "primary", tool: "fetch", arguments: { id: 1 } }, + { mcp: "backup", tool: "fetch", arguments: { id: 1 } } +]); + +// First complete (like Promise.race) - good for latency +const results = ctx.mcp.Race([ + { mcp: "region-us", tool: "ping", arguments: {} }, + { mcp: "region-eu", tool: "ping", arguments: {} } +]); + +// Access results +results.forEach(r => { + if (r.error) { + console.log(`${r.mcp}/${r.tool} failed: ${r.error}`); + } else { + console.log(`${r.mcp}/${r.tool} result:`, r.result); + } +}); +``` + ## Tool Schema Mapping Define input schemas for process transport tools: diff --git a/agent/llm/jsapi.go b/agent/llm/jsapi.go index f0697d1d..afd5ac82 100644 --- a/agent/llm/jsapi.go +++ b/agent/llm/jsapi.go @@ -674,6 +674,15 @@ func (api *JSAPI) executeAny(requests []*Request) []interface{} { errors = append(errors, &ir) } + // Drain remaining results in background (don't block) + if remaining > 0 { + go func(count int) { + for i := 0; i < count; i++ { + <-resultChan + } + }(remaining) + } + if firstSuccess != nil { return []interface{}{firstSuccess.result} } @@ -714,6 +723,17 @@ func (api *JSAPI) executeRace(requests []*Request) []interface{} { // Return first result result := <-resultChan + + // Drain remaining results in background (don't block) + remaining := len(requests) - 1 + if remaining > 0 { + go func(count int) { + for i := 0; i < count; i++ { + <-resultChan + } + }(remaining) + } + return []interface{}{result} } diff --git a/agent/memory/memory.go b/agent/memory/memory.go index 7a40d54a..b4db35b0 100644 --- a/agent/memory/memory.go +++ b/agent/memory/memory.go @@ -183,3 +183,57 @@ func (m *Memory) Clear() { m.Context.Clear() } } + +// Fork creates a new Memory instance with an independent Context namespace +// but sharing the User, Team, and Chat namespaces with the parent. +// This is used for parallel agent calls (ctx.agent.All/Any/Race) to prevent +// context state from being shared between concurrent sub-agent executions. +// +// The new Context namespace uses the provided newContextID. +// If newContextID is empty, returns a shallow copy with shared Context. +func (m *Memory) Fork(newContextID string) (*Memory, error) { + if m == nil { + return nil, nil + } + + // If no new context ID provided, share everything (shallow copy) + if newContextID == "" { + return &Memory{ + UserID: m.UserID, + TeamID: m.TeamID, + ChatID: m.ChatID, + ContextID: m.ContextID, + User: m.User, + Team: m.Team, + Chat: m.Chat, + Context: m.Context, + Config: m.Config, + }, nil + } + + // Create new Memory with independent Context namespace + forked := &Memory{ + UserID: m.UserID, + TeamID: m.TeamID, + ChatID: m.ChatID, + ContextID: newContextID, + User: m.User, // Shared + Team: m.Team, // Shared + Chat: m.Chat, // Shared + Context: nil, // Will be created below + Config: m.Config, + } + + // Create new Context namespace with independent ID + storeID := "" + if m.Config != nil { + storeID = m.Config.Context + } + ns, err := newNamespace(SpaceContext, newContextID, storeID, DefaultContextTTL) + if err != nil { + return nil, fmt.Errorf("failed to create forked context namespace: %w", err) + } + forked.Context = ns + + return forked, nil +}