Execute tool using stdio transport
(server_config: ServerConfig, tool_name: str, arguments: Dict[str, Any])
| 703 | return {"error": f"Error executing tool: {str(e)}"} |
| 704 | |
| 705 | async def execute_tool_stdio(server_config: ServerConfig, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: |
| 706 | """Execute tool using stdio transport""" |
| 707 | if not server_config.command: |
| 708 | return {"error": "stdio transport requires command"} |
| 709 | |
| 710 | # Find executable |
| 711 | full_command = find_executable(server_config.command) |
| 712 | if not full_command: |
| 713 | return {"error": f"Failed to find executable for command: {server_config.command}"} |
| 714 | |
| 715 | # Create environment with PATH included |
| 716 | merged_env = os.environ.copy() |
| 717 | if server_config.env: |
| 718 | merged_env.update(server_config.env) |
| 719 | |
| 720 | # Create server parameters |
| 721 | server_params = StdioServerParameters( |
| 722 | command=full_command, |
| 723 | args=server_config.args, |
| 724 | env=merged_env |
| 725 | ) |
| 726 | |
| 727 | logger.debug(f" Command: {full_command}") |
| 728 | logger.debug(f" Args: {server_config.args}") |
| 729 | |
| 730 | try: |
| 731 | # Use the MCP client with proper context management |
| 732 | async with stdio_client(server_params) as (read_stream, write_stream): |
| 733 | # Use our logging session |
| 734 | async with LoggingClientSession(read_stream, write_stream) as session: |
| 735 | return await execute_tool_with_session(session, tool_name, arguments) |
| 736 | |
| 737 | except Exception as e: |
| 738 | logger.error(f"Error with stdio tool execution: {e}") |
| 739 | logger.error(traceback.format_exc()) |
| 740 | return {"error": f"Error executing tool via stdio: {str(e)}"} |
| 741 | |
| 742 | async def execute_tool_sse(server_config: ServerConfig, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: |
| 743 | """Execute tool using SSE transport""" |