call_tool
Execute tools on MCP servers by name with arguments, returning results with timing and metadata for testing and validation purposes.
Instructions
Execute a tool on the connected MCP server.
Calls a tool by name with the provided arguments and returns the result along with execution timing and metadata.
Returns: Dictionary with tool execution results including: - success: True if tool executed successfully - tool_call: Object with tool_name, arguments, result, and execution metadata - metadata: Request timing and server information
Raises: Returns error dict for various failure scenarios: - not_connected: No active connection - tool_not_found: Tool doesn't exist on server - invalid_arguments: Arguments don't match tool schema - execution_error: Tool execution failed
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| name | Yes | Name of the tool to execute on the target MCP server | |
| arguments | Yes | Dictionary of arguments to pass to the tool |
Implementation Reference
- The core handler implementation for the 'call_tool' MCP tool. Decorated with @mcp.tool for automatic registration. Proxies tool execution to a connected target MCP server using a client instance from ConnectionManager, processes the result, handles various errors (connection, tool not found, invalid args, execution), provides detailed logging and user notifications via ctx, and returns structured results with success status, tool_call details, and metadata including timings and stats.
@mcp.tool async def call_tool( name: Annotated[str, "Name of the tool to execute on the target MCP server"], arguments: Annotated[dict[str, Any], "Dictionary of arguments to pass to the tool"], ctx: Context ) -> dict[str, Any]: """Execute a tool on the connected MCP server. Calls a tool by name with the provided arguments and returns the result along with execution timing and metadata. Returns: Dictionary with tool execution results including: - success: True if tool executed successfully - tool_call: Object with tool_name, arguments, result, and execution metadata - metadata: Request timing and server information Raises: Returns error dict for various failure scenarios: - not_connected: No active connection - tool_not_found: Tool doesn't exist on server - invalid_arguments: Arguments don't match tool schema - execution_error: Tool execution failed """ start_time = time.perf_counter() try: # Verify connection exists client, state = ConnectionManager.require_connection() # User-facing progress update await ctx.info(f"Calling tool '{name}' on target server") # Detailed technical log logger.info( f"Calling tool '{name}' with arguments", extra={"tool_name": name, "arguments": arguments}, ) # Execute the tool tool_start = time.perf_counter() result = await client.call_tool(name, arguments) tool_elapsed_ms = (time.perf_counter() - tool_start) * 1000 # Increment statistics ConnectionManager.increment_stat("tools_called") total_elapsed_ms = (time.perf_counter() - start_time) * 1000 # Extract result content result_content = None if hasattr(result, "content") and result.content: # Handle list of content items if isinstance(result.content, list) and len(result.content) > 0: content_item = result.content[0] if hasattr(content_item, "text"): result_content = content_item.text elif hasattr(content_item, "data"): result_content = content_item.data else: result_content = str(content_item) else: result_content = result.content elif hasattr(result, "result"): result_content = result.result else: result_content = str(result) # User-facing success update await ctx.info(f"Tool '{name}' executed successfully") # Detailed technical log logger.info( f"Tool '{name}' executed successfully", extra={ "tool_name": name, "execution_ms": tool_elapsed_ms, "total_ms": total_elapsed_ms, }, ) return { "success": True, "tool_call": { "tool_name": name, "arguments": arguments, "result": result_content, "execution": { "duration_ms": round(tool_elapsed_ms, 2), "success": True, }, }, "metadata": { "request_time_ms": round(total_elapsed_ms, 2), "server_url": state.server_url, "connection_statistics": state.statistics, }, } except ConnectionError as e: elapsed_ms = (time.perf_counter() - start_time) * 1000 # User-facing error update await ctx.error(f"Not connected when calling tool '{name}': {str(e)}") # Detailed technical log logger.error( f"Not connected when calling tool '{name}': {str(e)}", extra={"tool_name": name, "duration_ms": elapsed_ms}, ) return { "success": False, "error": { "error_type": "not_connected", "message": str(e), "details": {"tool_name": name}, "suggestion": "Use connect_to_server() to establish a connection first", }, "tool_call": None, "metadata": { "request_time_ms": round(elapsed_ms, 2), }, } except Exception as e: elapsed_ms = (time.perf_counter() - start_time) * 1000 # Determine error type based on exception message error_type = "execution_error" suggestion = "Check the tool name and arguments, then retry" error_msg = str(e).lower() if "not found" in error_msg or "unknown tool" in error_msg: error_type = "tool_not_found" suggestion = f"Tool '{name}' does not exist on the server. Use list_tools() to see available tools" elif "argument" in error_msg or "parameter" in error_msg or "validation" in error_msg: error_type = "invalid_arguments" suggestion = f"Arguments do not match the tool schema. Use list_tools() to see the correct schema for '{name}'" # User-facing error update await ctx.error(f"Failed to call tool '{name}': {str(e)}") # Detailed technical log logger.error( f"Failed to call tool '{name}': {str(e)}", extra={ "tool_name": name, "arguments": arguments, "error_type": error_type, "duration_ms": elapsed_ms, }, ) # Increment error counter ConnectionManager.increment_stat("errors") return { "success": False, "error": { "error_type": error_type, "message": f"Failed to call tool '{name}': {str(e)}", "details": { "tool_name": name, "arguments": arguments, "exception_type": type(e).__name__, }, "suggestion": suggestion, }, "tool_call": None, "metadata": { "request_time_ms": round(elapsed_ms, 2), }, } - Input schema defined via Annotated type hints on function parameters, which FastMCP uses to generate JSON schema for the tool.
name: Annotated[str, "Name of the tool to execute on the target MCP server"], arguments: Annotated[dict[str, Any], "Dictionary of arguments to pass to the tool"], ctx: Context ) -> dict[str, Any]: - node-wrapper/python-src/src/mcp_test_mcp/server.py:98-98 (registration)Import of the 'tools' module (containing call_tool) in the main server.py, which triggers the execution of module-level code including @mcp.tool decorators that register the tool with the shared FastMCP 'mcp' instance.
from .tools import connection, tools, resources, prompts, llm - Exported in __all__ of tools/__init__.py, making call_tool available when the package is imported.
"call_tool",