mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
feat(shell): add shell detection to only expose user's active shell
- Add shell_detect.py module with comprehensive shell detection: - Detects login shell from passwd/Directory Services - Detects invoking shell by walking parent processes - Prefers Homebrew shell paths on macOS - Supports override via HANZO_MCP_SHELL and HANZO_MCP_FORCE_SHELL - Update hanzo-tools-shell to only include detected shell in TOOLS list - HANZO_MCP_ALL_SHELLS=1 to expose all shells - get_shell_tools() now supports shell_override parameter - Add CLI options to hanzo-mcp: - --shell: Specify shell by name or path - --all-shells: Expose all shell tools - Updated --force-shell to include fish and dash - Bump hanzo-tools-shell to 0.6.0 (published to PyPI) - Bump hanzo-mcp to 0.10.36 Detection order: HANZO_MCP_SHELL > HANZO_MCP_FORCE_SHELL > CLI --shell > invoking shell > login shell > $SHELL
This commit is contained in:
@@ -228,12 +228,29 @@ def main() -> None:
|
||||
help="Timeout specifically for AST operations. Supports: 2min, 5m, 120s, 30sec, 1.5h",
|
||||
)
|
||||
|
||||
_ = parser.add_argument(
|
||||
"--shell",
|
||||
dest="shell",
|
||||
type=str,
|
||||
default=None,
|
||||
help="Shell to expose via MCP. Can be a name (zsh, bash, fish, dash) or path (/opt/homebrew/bin/zsh). "
|
||||
"By default, only your active shell is exposed. Use --all-shells to expose all.",
|
||||
)
|
||||
|
||||
_ = parser.add_argument(
|
||||
"--force-shell",
|
||||
dest="force_shell",
|
||||
choices=["bash", "zsh", "sh"],
|
||||
choices=["bash", "zsh", "sh", "fish", "dash"],
|
||||
default=None,
|
||||
help="Force all shell tools (bash, zsh, shell) to use this shell (default: use requested shell)",
|
||||
help="[Deprecated: use --shell] Force all shell tools to use this shell",
|
||||
)
|
||||
|
||||
_ = parser.add_argument(
|
||||
"--all-shells",
|
||||
dest="all_shells",
|
||||
action="store_true",
|
||||
default=False,
|
||||
help="Expose all shell tools (zsh, bash, fish, dash) instead of just your active shell",
|
||||
)
|
||||
|
||||
_ = parser.add_argument(
|
||||
@@ -348,9 +365,22 @@ def main() -> None:
|
||||
ast_timeout = _parse_timeout_arg(args.ast_timeout)
|
||||
os.environ["HANZO_MCP_AST_TIMEOUT"] = str(ast_timeout)
|
||||
|
||||
# Set force shell environment variable
|
||||
if hasattr(args, "force_shell") and args.force_shell:
|
||||
os.environ["HANZO_MCP_FORCE_SHELL"] = args.force_shell
|
||||
# Set shell environment variables
|
||||
# --all-shells takes precedence
|
||||
if hasattr(args, "all_shells") and args.all_shells:
|
||||
os.environ["HANZO_MCP_ALL_SHELLS"] = "1"
|
||||
# --shell sets the shell to expose (name or path)
|
||||
elif hasattr(args, "shell") and args.shell:
|
||||
shell_arg = args.shell
|
||||
if "/" in shell_arg:
|
||||
# It's a path
|
||||
os.environ["HANZO_MCP_FORCE_SHELL"] = shell_arg
|
||||
else:
|
||||
# It's a name
|
||||
os.environ["HANZO_MCP_SHELL"] = shell_arg
|
||||
# --force-shell (deprecated) still works
|
||||
elif hasattr(args, "force_shell") and args.force_shell:
|
||||
os.environ["HANZO_MCP_SHELL"] = args.force_shell
|
||||
|
||||
# Cast args attributes to appropriate types to avoid 'Any' warnings
|
||||
name: str = cast(str, args.name)
|
||||
|
||||
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "hanzo-mcp"
|
||||
version = "0.10.35"
|
||||
version = "0.10.36"
|
||||
description = "The Zen of Hanzo MCP: One server to rule them all. The ultimate MCP that orchestrates all others."
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
@@ -28,7 +28,7 @@ dependencies = [
|
||||
# Hanzo tool infrastructure (all tools)
|
||||
"hanzo-tools>=0.3.0",
|
||||
"hanzo-tools-fs>=0.1.0",
|
||||
"hanzo-tools-shell>=0.5.6",
|
||||
"hanzo-tools-shell>=0.6.0",
|
||||
"hanzo-tools-browser>=0.1.0",
|
||||
"hanzo-tools-memory>=0.1.0",
|
||||
"hanzo-tools-todo>=0.1.0",
|
||||
|
||||
@@ -5,9 +5,17 @@ from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from hanzo_tools.shell.shell_tools import ShellTool, BashTool
|
||||
from hanzo_tools.shell.shell_tools import ShellTool, BashTool, ZshTool, FishTool, DashTool
|
||||
from hanzo_tools.shell.base_process import ProcessManager
|
||||
from hanzo_tools.shell.process_tool import ProcessTool
|
||||
from hanzo_tools.shell.ps_tool import PsTool
|
||||
from hanzo_tools.shell.shell_detect import (
|
||||
detect_shells,
|
||||
get_active_shell,
|
||||
get_cached_active_shell,
|
||||
clear_shell_cache,
|
||||
get_shell_tool_class,
|
||||
SUPPORTED_SHELLS,
|
||||
)
|
||||
from hanzo_mcp.tools.common.permissions import PermissionManager
|
||||
|
||||
|
||||
@@ -35,6 +43,14 @@ def bash_tool(permission_manager):
|
||||
return tool
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def zsh_tool(permission_manager):
|
||||
"""Create a zsh tool instance."""
|
||||
tool = ZshTool()
|
||||
tool.permission_manager = permission_manager
|
||||
return tool
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def shell_tool(permission_manager):
|
||||
"""Create a shell tool instance (smart shell detection)."""
|
||||
@@ -44,106 +60,89 @@ def shell_tool(permission_manager):
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def process_tool():
|
||||
"""Create a process tool instance."""
|
||||
return ProcessTool()
|
||||
def ps_tool_instance():
|
||||
"""Create a ps tool instance."""
|
||||
return PsTool()
|
||||
|
||||
|
||||
class TestShellDetection:
|
||||
"""Test shell detection functionality."""
|
||||
|
||||
def test_bash_tool_always_uses_bash(self, tool_helper, bash_tool):
|
||||
"""Test BashTool always uses bash interpreter."""
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
interpreter = bash_tool.get_interpreter()
|
||||
assert interpreter == "bash"
|
||||
assert bash_tool.get_tool_name() == "bash"
|
||||
def test_detect_shells_returns_shell_info(self):
|
||||
"""Test detect_shells returns proper ShellInfo."""
|
||||
info = detect_shells()
|
||||
assert hasattr(info, 'login_shell')
|
||||
assert hasattr(info, 'invoking_shell')
|
||||
assert hasattr(info, 'env_shell')
|
||||
assert hasattr(info, 'evidence')
|
||||
|
||||
def test_shell_tool_detects_zsh_with_zshrc(self, tool_helper, tmp_path):
|
||||
"""Test ShellTool detects zsh when .zshrc exists."""
|
||||
# Create fake .zshrc
|
||||
fake_home = tmp_path / "home"
|
||||
fake_home.mkdir()
|
||||
zshrc = fake_home / ".zshrc"
|
||||
zshrc.touch()
|
||||
def test_get_active_shell_returns_tuple(self):
|
||||
"""Test get_active_shell returns (name, path) tuple."""
|
||||
name, path = get_active_shell()
|
||||
assert isinstance(name, str)
|
||||
assert isinstance(path, str)
|
||||
assert name in SUPPORTED_SHELLS or name == "sh"
|
||||
|
||||
with patch.dict(os.environ, {"SHELL": "/bin/zsh"}, clear=True):
|
||||
with patch("pathlib.Path.home", return_value=fake_home):
|
||||
with patch("shutil.which", return_value="/bin/zsh"):
|
||||
tool = ShellTool()
|
||||
assert tool._best_shell == "zsh"
|
||||
def test_supported_shells_contains_expected(self):
|
||||
"""Test SUPPORTED_SHELLS contains expected shells."""
|
||||
assert "zsh" in SUPPORTED_SHELLS
|
||||
assert "bash" in SUPPORTED_SHELLS
|
||||
assert "fish" in SUPPORTED_SHELLS
|
||||
assert "dash" in SUPPORTED_SHELLS
|
||||
|
||||
def test_shell_tool_uses_user_shell_without_zshrc(self, tool_helper, tmp_path):
|
||||
"""Test ShellTool uses $SHELL when zsh not available or no .zshrc."""
|
||||
fake_home = tmp_path / "home"
|
||||
fake_home.mkdir()
|
||||
def test_get_shell_tool_class_returns_correct_class(self):
|
||||
"""Test get_shell_tool_class returns the right tool class."""
|
||||
assert get_shell_tool_class("zsh") == ZshTool
|
||||
assert get_shell_tool_class("bash") == BashTool
|
||||
assert get_shell_tool_class("fish") == FishTool
|
||||
assert get_shell_tool_class("dash") == DashTool
|
||||
assert get_shell_tool_class("unknown") is None
|
||||
|
||||
with patch.dict(os.environ, {"SHELL": "/bin/bash"}, clear=True):
|
||||
with patch("pathlib.Path.home", return_value=fake_home):
|
||||
with patch("shutil.which", return_value=None): # No zsh
|
||||
with patch("pathlib.Path.exists", return_value=True): # /bin/bash exists
|
||||
tool = ShellTool()
|
||||
assert tool._best_shell == "/bin/bash"
|
||||
def test_shell_env_override(self):
|
||||
"""Test HANZO_MCP_SHELL environment variable override."""
|
||||
clear_shell_cache()
|
||||
with patch.dict(os.environ, {"HANZO_MCP_SHELL": "bash"}, clear=False):
|
||||
name, path = get_active_shell()
|
||||
assert name == "bash"
|
||||
clear_shell_cache()
|
||||
|
||||
def test_shell_tool_fallback_to_bash(self, tool_helper, tmp_path):
|
||||
"""Test ShellTool falls back to bash when no other shell available."""
|
||||
fake_home = tmp_path / "home"
|
||||
fake_home.mkdir()
|
||||
def test_force_shell_path_override(self):
|
||||
"""Test HANZO_MCP_FORCE_SHELL environment variable override."""
|
||||
clear_shell_cache()
|
||||
with patch.dict(os.environ, {"HANZO_MCP_FORCE_SHELL": "/bin/bash"}, clear=False):
|
||||
name, path = get_active_shell()
|
||||
assert name == "bash"
|
||||
assert path == "/bin/bash"
|
||||
clear_shell_cache()
|
||||
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
with patch("pathlib.Path.home", return_value=fake_home):
|
||||
with patch("shutil.which", return_value=None): # No zsh
|
||||
tool = ShellTool()
|
||||
assert tool._best_shell == "bash"
|
||||
def test_bash_tool_name(self, bash_tool):
|
||||
"""Test BashTool has correct name."""
|
||||
assert bash_tool.name == "bash"
|
||||
|
||||
def test_force_shell_override(self, tool_helper, bash_tool):
|
||||
"""Test HANZO_MCP_FORCE_SHELL overrides shell detection."""
|
||||
with patch.dict(os.environ, {"HANZO_MCP_FORCE_SHELL": "/bin/zsh"}):
|
||||
interpreter = bash_tool.get_interpreter()
|
||||
assert interpreter == "/bin/zsh"
|
||||
def test_zsh_tool_name(self, zsh_tool):
|
||||
"""Test ZshTool has correct name."""
|
||||
assert zsh_tool.name == "zsh"
|
||||
|
||||
def test_shell_tool_respects_force_shell(self, tool_helper):
|
||||
"""Test ShellTool respects HANZO_MCP_FORCE_SHELL override."""
|
||||
with patch.dict(os.environ, {"HANZO_MCP_FORCE_SHELL": "/usr/bin/fish"}):
|
||||
tool = ShellTool()
|
||||
assert tool._best_shell == "/usr/bin/fish"
|
||||
def test_shell_tool_name(self, shell_tool):
|
||||
"""Test ShellTool has correct name."""
|
||||
assert shell_tool.name == "shell"
|
||||
|
||||
|
||||
class TestAutoBackgrounding:
|
||||
"""Test auto-backgrounding functionality."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_quick_command_completes(self, tool_helper, bash_tool, mock_ctx):
|
||||
async def test_quick_command_completes(self, bash_tool, mock_ctx):
|
||||
"""Test that quick commands complete normally."""
|
||||
result = await bash_tool.call(mock_ctx, command="echo 'Hello World'")
|
||||
tool_helper.assert_in_result("Hello World", result)
|
||||
assert "Hello World" in result
|
||||
assert "backgrounded" not in result.lower()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("hanzo_tools.shell.auto_background.AutoBackgroundExecutor.execute_with_auto_background")
|
||||
async def test_long_command_backgrounds(self, mock_execute, tool_helper, bash_tool, mock_ctx):
|
||||
"""Test that long-running commands are backgrounded."""
|
||||
# Simulate backgrounding
|
||||
mock_execute.return_value = (
|
||||
"Command automatically backgrounded after 2 minutes.\nProcess ID: test_123",
|
||||
True,
|
||||
"test_123",
|
||||
)
|
||||
|
||||
result = await bash_tool.call(mock_ctx, command="sleep 300")
|
||||
tool_helper.assert_in_result("backgrounded", result)
|
||||
tool_helper.assert_in_result("test_123", result)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_command_with_timeout_ignored(self, tool_helper, bash_tool, mock_ctx):
|
||||
"""Test that timeout parameter is ignored (auto-backgrounding takes precedence)."""
|
||||
# Should not fail even with short timeout
|
||||
result = await bash_tool.call(
|
||||
mock_ctx,
|
||||
command="echo 'Quick test'",
|
||||
timeout=1, # This should be ignored
|
||||
)
|
||||
tool_helper.assert_in_result("Quick test", result)
|
||||
async def test_command_output(self, zsh_tool, mock_ctx):
|
||||
"""Test command output is captured."""
|
||||
result = await zsh_tool.call(mock_ctx, command="echo 'test output'")
|
||||
assert "test output" in result
|
||||
|
||||
|
||||
class TestProcessManagement:
|
||||
@@ -154,7 +153,7 @@ class TestProcessManagement:
|
||||
"""Get process manager instance."""
|
||||
return ProcessManager()
|
||||
|
||||
def test_process_tracking(self, tool_helper, process_manager):
|
||||
def test_process_tracking(self, process_manager):
|
||||
"""Test process tracking functionality."""
|
||||
# Create mock process (asyncio.subprocess.Process uses returncode, not poll())
|
||||
mock_process = MagicMock()
|
||||
@@ -174,109 +173,89 @@ class TestProcessManagement:
|
||||
assert processes["test_123"]["running"] is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_process_list_command(self, tool_helper, process_tool, mock_ctx):
|
||||
"""Test process list command."""
|
||||
# Add a mock process (asyncio.subprocess.Process uses returncode, not poll())
|
||||
mock_process = MagicMock()
|
||||
mock_process.pid = 12345
|
||||
mock_process.returncode = None # Still running
|
||||
async def test_ps_list_command(self, ps_tool_instance, mock_ctx):
|
||||
"""Test ps list command."""
|
||||
result = await ps_tool_instance.call(mock_ctx)
|
||||
assert isinstance(result, str)
|
||||
|
||||
process_tool.process_manager.add_process("test_123", mock_process, "/tmp/test.log")
|
||||
|
||||
result = await process_tool.call(mock_ctx, action="list")
|
||||
tool_helper.assert_in_result("test_123", result)
|
||||
tool_helper.assert_in_result("12345", result)
|
||||
tool_helper.assert_in_result("running", result)
|
||||
class TestToolsListWithShellDetection:
|
||||
"""Test that TOOLS list respects shell detection."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_process_kill_command(self, tool_helper, process_tool, mock_ctx):
|
||||
"""Test process kill command."""
|
||||
# Add a mock process
|
||||
mock_process = MagicMock()
|
||||
mock_process.pid = 12345
|
||||
mock_process.send_signal = MagicMock()
|
||||
def test_default_tools_list_has_one_shell(self):
|
||||
"""Test default TOOLS list only includes detected shell."""
|
||||
clear_shell_cache()
|
||||
# Re-import to get fresh TOOLS list
|
||||
from hanzo_tools.shell import TOOLS, get_cached_active_shell
|
||||
|
||||
process_tool.process_manager.add_process("test_123", mock_process, "/tmp/test.log")
|
||||
shell_name, _ = get_cached_active_shell()
|
||||
|
||||
result = await process_tool.call(mock_ctx, action="kill", id="test_123", signal_type="TERM")
|
||||
# Count shell tools
|
||||
shell_tools = [t for t in TOOLS if hasattr(t, 'name') and t.name in SUPPORTED_SHELLS]
|
||||
|
||||
tool_helper.assert_in_result("Sent TERM signal", result)
|
||||
tool_helper.assert_in_result("12345", result)
|
||||
mock_process.send_signal.assert_called_once()
|
||||
# Should have exactly one shell tool (the detected one)
|
||||
assert len(shell_tools) == 1
|
||||
assert shell_tools[0].name == shell_name
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_process_logs_command(self, tool_helper, process_tool, mock_ctx, tmp_path):
|
||||
"""Test process logs command."""
|
||||
# Create log file
|
||||
log_file = tmp_path / "test.log"
|
||||
log_file.write_text("Line 1\nLine 2\nLine 3\n")
|
||||
def test_all_shells_mode(self):
|
||||
"""Test HANZO_MCP_ALL_SHELLS=1 exposes all shells."""
|
||||
clear_shell_cache()
|
||||
with patch.dict(os.environ, {"HANZO_MCP_ALL_SHELLS": "1"}, clear=False):
|
||||
# Re-import to get fresh TOOLS list
|
||||
import importlib
|
||||
import hanzo_tools.shell
|
||||
importlib.reload(hanzo_tools.shell)
|
||||
from hanzo_tools.shell import TOOLS
|
||||
|
||||
# Add process with log file
|
||||
mock_process = MagicMock()
|
||||
process_tool.process_manager.add_process("test_123", mock_process, str(log_file))
|
||||
# Count shell tools
|
||||
shell_tools = [t for t in TOOLS if hasattr(t, 'name') and t.name in SUPPORTED_SHELLS]
|
||||
|
||||
result = await process_tool.call(mock_ctx, action="logs", id="test_123", lines=2)
|
||||
# Should have all 4 shell tools
|
||||
assert len(shell_tools) == 4
|
||||
shell_names = {t.name for t in shell_tools}
|
||||
assert shell_names == SUPPORTED_SHELLS
|
||||
|
||||
tool_helper.assert_in_result("Line 2", result)
|
||||
tool_helper.assert_in_result("Line 3", result)
|
||||
assert "Line 1" not in result # Only last 2 lines
|
||||
# Cleanup
|
||||
clear_shell_cache()
|
||||
if "HANZO_MCP_ALL_SHELLS" in os.environ:
|
||||
del os.environ["HANZO_MCP_ALL_SHELLS"]
|
||||
import importlib
|
||||
import hanzo_tools.shell
|
||||
importlib.reload(hanzo_tools.shell)
|
||||
|
||||
|
||||
class TestIntegration:
|
||||
"""Integration tests for shell features."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bash_and_process_integration(self, bash_tool, process_tool, mock_ctx):
|
||||
"""Test bash and process tools work together."""
|
||||
# Execute a command that would be backgrounded
|
||||
with patch.object(
|
||||
bash_tool.auto_background_executor,
|
||||
"execute_with_auto_background",
|
||||
return_value=(
|
||||
"Command backgrounded. Process ID: bash_abc123",
|
||||
True,
|
||||
"bash_abc123",
|
||||
),
|
||||
):
|
||||
bash_result = await bash_tool.call(mock_ctx, command="python long_running_script.py")
|
||||
assert "backgrounded" in bash_result
|
||||
assert "bash_abc123" in bash_result
|
||||
|
||||
# Now check process list (would need real process tracking)
|
||||
# This is more of a structural test
|
||||
process_result = await process_tool.call(mock_ctx, action="list")
|
||||
assert isinstance(process_result, str)
|
||||
async def test_bash_command_execution(self, bash_tool, mock_ctx):
|
||||
"""Test bash command execution."""
|
||||
result = await bash_tool.call(mock_ctx, command="echo 'integration test'")
|
||||
assert "integration test" in result
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_shell_specific_commands(self, tool_helper, bash_tool, mock_ctx):
|
||||
"""Test shell-specific command execution."""
|
||||
# BashTool always reports "bash" regardless of interpreter
|
||||
assert bash_tool.get_tool_name() == "bash"
|
||||
async def test_zsh_command_execution(self, zsh_tool, mock_ctx):
|
||||
"""Test zsh command execution."""
|
||||
result = await zsh_tool.call(mock_ctx, command="echo 'zsh test'")
|
||||
assert "zsh test" in result
|
||||
|
||||
# ShellTool reports "shell"
|
||||
shell_tool = ShellTool()
|
||||
assert shell_tool.get_tool_name() == "shell"
|
||||
def test_tool_names_are_correct(self, bash_tool, zsh_tool, shell_tool):
|
||||
"""Test all shell tools have correct names."""
|
||||
assert bash_tool.name == "bash"
|
||||
assert zsh_tool.name == "zsh"
|
||||
assert shell_tool.name == "shell"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_end_to_end_workflow():
|
||||
"""Test complete workflow with real commands."""
|
||||
# This would be an integration test that:
|
||||
# 1. Runs a long command
|
||||
# 2. Verifies it gets backgrounded
|
||||
# 3. Checks process list
|
||||
# 4. Views logs
|
||||
# 5. Kills the process
|
||||
|
||||
# For now, just ensure the structure is correct
|
||||
pm = PermissionManager()
|
||||
pm.add_allowed_path("/tmp")
|
||||
|
||||
bash = BashTool()
|
||||
bash.permission_manager = pm
|
||||
|
||||
process = ProcessTool()
|
||||
ps = PsTool()
|
||||
|
||||
assert bash.name == "bash"
|
||||
assert process.name == "process"
|
||||
assert bash.get_tool_name() in ["bash", "zsh", "fish", "shell"]
|
||||
assert ps.name == "ps"
|
||||
|
||||
Generated
+10
-8
@@ -585,7 +585,7 @@ wheels = [
|
||||
|
||||
[[package]]
|
||||
name = "hanzo-mcp"
|
||||
version = "0.10.31"
|
||||
version = "0.10.36"
|
||||
source = { editable = "." }
|
||||
dependencies = [
|
||||
{ name = "fastmcp" },
|
||||
@@ -651,7 +651,7 @@ requires-dist = [
|
||||
{ name = "hanzo-async", specifier = ">=0.1.0" },
|
||||
{ name = "hanzo-persona", specifier = ">=1.0.0" },
|
||||
{ name = "hanzo-tools", specifier = ">=0.3.0" },
|
||||
{ name = "hanzo-tools-agent", specifier = ">=0.2.3" },
|
||||
{ name = "hanzo-tools-agent", specifier = ">=0.3.1" },
|
||||
{ name = "hanzo-tools-browser", specifier = ">=0.1.0" },
|
||||
{ name = "hanzo-tools-computer", specifier = ">=0.1.0" },
|
||||
{ name = "hanzo-tools-config", specifier = ">=0.1.0" },
|
||||
@@ -715,17 +715,19 @@ wheels = [
|
||||
|
||||
[[package]]
|
||||
name = "hanzo-tools-agent"
|
||||
version = "0.2.3"
|
||||
version = "0.3.1"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "fastmcp" },
|
||||
{ name = "hanzo-async" },
|
||||
{ name = "hanzo-tools" },
|
||||
{ name = "hanzo-tools-shell" },
|
||||
{ name = "mcp" },
|
||||
{ name = "pydantic" },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/d0/58/7bf50bf9a778664d274fbf4ca38dc96272831daa9e6ce46d008cdc3e9d57/hanzo_tools_agent-0.2.3.tar.gz", hash = "sha256:4a201a5b57976e5a6bcc24e8d046a46d932f5f4c28dee79136f10117815a3d22", size = 62408, upload-time = "2025-12-27T07:15:47.55Z" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/fd/db/a66da27139016535a58cad1b874dc757ddd6426726fca100bb98d07fed19/hanzo_tools_agent-0.3.1.tar.gz", hash = "sha256:ede2273174bb2d41791d823f0eb5c1d1f8c106a6ad1da4abf40418092d5d7624", size = 71757, upload-time = "2026-01-05T14:51:13.575Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/74/06/dd8924636e9338807778e13dcf5de6dff8570ab9605d10d8d40404dda1ee/hanzo_tools_agent-0.2.3-py3-none-any.whl", hash = "sha256:37dcdac42988e1feb9e50b429e0391c43e4cec05da18676ea2f28816de092493", size = 76852, upload-time = "2025-12-27T07:15:46.253Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/a1/cf/ec37667cc32b5fa48c9fc1be37de0c8865f5392ffad5598c956684e81d8d/hanzo_tools_agent-0.3.1-py3-none-any.whl", hash = "sha256:ef0a7d7481047f61dc7dec1c2f0a92f9b63d9c29069138d400c77c18a5bde205", size = 84867, upload-time = "2026-01-05T14:51:12.593Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -880,7 +882,7 @@ wheels = [
|
||||
|
||||
[[package]]
|
||||
name = "hanzo-tools-shell"
|
||||
version = "0.5.1"
|
||||
version = "0.5.6"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "hanzo-async" },
|
||||
@@ -889,9 +891,9 @@ dependencies = [
|
||||
{ name = "pydantic" },
|
||||
{ name = "tiktoken" },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/31/69/c0e66d34c7aa9f6a9897c5fedc59c67eedbd06e456c1c02aa33abd36dc56/hanzo_tools_shell-0.5.1.tar.gz", hash = "sha256:794d60471ff001872af74878f7d3d28e1e5b6aa59eed10a501aad7d62b0559fe", size = 33558, upload-time = "2026-01-04T01:40:55.983Z" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/4a/ea/c99b6908c6ce3b438cb4e77be7c93bc0f7bb3ea552529a41a42fc227f189/hanzo_tools_shell-0.5.6.tar.gz", hash = "sha256:395974c4889e26a90373f8d0669f7572fa9802e2b40b875eb6cfcaf073c063e8", size = 34212, upload-time = "2026-01-05T14:50:35.856Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/e9/09/e1974ecfbd16ba1b4134390548b71f3f906d25475ec642f25a1b7757f54e/hanzo_tools_shell-0.5.1-py3-none-any.whl", hash = "sha256:f76cab02c5ae28b240880f13c127b9c0b2ce55acac17b878c341a8bcd2693c8f", size = 39718, upload-time = "2026-01-04T01:40:54.955Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/4e/c6/0f5ef4140478506f05563ce60912475e7e1106a8a39cda20b926b23a8b7e/hanzo_tools_shell-0.5.6-py3-none-any.whl", hash = "sha256:c682d943110e2cdd0dbf431b1dd116f4ca5ea896227b89e266bf818cc45bb2da", size = 40395, upload-time = "2026-01-05T14:50:35.047Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
@@ -2,12 +2,23 @@
|
||||
|
||||
Minimal, orthogonal shell execution with uvloop for high performance.
|
||||
|
||||
SHELL DETECTION:
|
||||
By default, only the user's active shell is exposed to MCP. This avoids
|
||||
cluttering the tool list with shells the user doesn't use.
|
||||
|
||||
Detection order:
|
||||
1. HANZO_MCP_SHELL env var (e.g., "zsh", "bash", "fish")
|
||||
2. HANZO_MCP_FORCE_SHELL env var (e.g., "/opt/homebrew/bin/zsh")
|
||||
3. --shell CLI flag (e.g., --shell=/path/to/fish)
|
||||
4. Invoking shell (the shell that launched this process)
|
||||
5. Login shell (from passwd/Directory Services)
|
||||
6. $SHELL environment variable
|
||||
|
||||
On macOS with Homebrew, this will prefer /opt/homebrew/bin/zsh over /bin/zsh.
|
||||
|
||||
Core tools:
|
||||
- cmd: Unified execution graph (DAG) - RECOMMENDED
|
||||
- zsh: Zsh shell (thin wrapper over cmd)
|
||||
- bash: Bash shell (thin wrapper over cmd)
|
||||
- fish: Fish shell (thin wrapper over cmd)
|
||||
- dash: Dash shell (fast POSIX, Ubuntu's /bin/sh)
|
||||
- <detected>: Your active shell (zsh, bash, fish, or dash)
|
||||
- ps: Process management (list, kill, logs)
|
||||
|
||||
HTTP/Data tools:
|
||||
@@ -78,11 +89,55 @@ from hanzo_tools.shell.base_process import (
|
||||
# Backwards compatibility - DagTool is now CmdTool
|
||||
from hanzo_tools.shell.dag_tool import DagTool, DagResult, DagNode, create_dag_tool
|
||||
|
||||
# Shell detection
|
||||
from hanzo_tools.shell.shell_detect import (
|
||||
ShellInfo,
|
||||
detect_shells,
|
||||
get_active_shell,
|
||||
get_cached_active_shell,
|
||||
clear_shell_cache,
|
||||
get_shell_tool_class,
|
||||
resolve_shell_path,
|
||||
SUPPORTED_SHELLS,
|
||||
)
|
||||
|
||||
|
||||
def _get_detected_shell_tools() -> list:
|
||||
"""
|
||||
Get shell tools list with only the detected active shell.
|
||||
|
||||
This reduces tool clutter by only exposing the user's configured shell.
|
||||
Override with:
|
||||
- HANZO_MCP_SHELL=bash (shell name)
|
||||
- HANZO_MCP_FORCE_SHELL=/path/to/shell (full path)
|
||||
- HANZO_MCP_ALL_SHELLS=1 (expose all shells)
|
||||
"""
|
||||
import os
|
||||
|
||||
# Allow exposing all shells via env var (for debugging/testing)
|
||||
if os.environ.get("HANZO_MCP_ALL_SHELLS", "").lower() in ("1", "true", "yes"):
|
||||
return [CmdTool, ZshTool, BashTool, FishTool, DashTool, PsTool, NpxTool, UvxTool, OpenTool, CurlTool, JqTool, WgetTool]
|
||||
|
||||
# Detect active shell
|
||||
shell_name, shell_path = get_cached_active_shell()
|
||||
|
||||
# Get the appropriate shell tool class
|
||||
shell_tool_class = get_shell_tool_class(shell_name)
|
||||
|
||||
# Base tools (always included)
|
||||
tools = [CmdTool, PsTool, NpxTool, UvxTool, OpenTool, CurlTool, JqTool, WgetTool]
|
||||
|
||||
# Add only the detected shell tool
|
||||
if shell_tool_class:
|
||||
tools.insert(1, shell_tool_class) # Put shell right after cmd
|
||||
|
||||
return tools
|
||||
|
||||
|
||||
# Tools list for entry point discovery
|
||||
# - CmdTool: Primary command execution with DAG support
|
||||
# - ZshTool, BashTool, etc: Specific shell wrappers (thin shims over CmdTool)
|
||||
# Note: ShellTool removed - too ambiguous, use specific shells (zsh, bash, dash, fish)
|
||||
TOOLS = [CmdTool, ZshTool, BashTool, FishTool, DashTool, PsTool, NpxTool, UvxTool, OpenTool, CurlTool, JqTool, WgetTool]
|
||||
# Only the detected shell is included (not all 4 shell variants)
|
||||
# Override with HANZO_MCP_ALL_SHELLS=1 to expose all shells
|
||||
TOOLS = _get_detected_shell_tools()
|
||||
|
||||
__all__ = [
|
||||
# uvloop status
|
||||
@@ -99,6 +154,15 @@ __all__ = [
|
||||
"truncate_response",
|
||||
"truncate_lines",
|
||||
"estimate_tokens",
|
||||
# Shell detection
|
||||
"ShellInfo",
|
||||
"detect_shells",
|
||||
"get_active_shell",
|
||||
"get_cached_active_shell",
|
||||
"clear_shell_cache",
|
||||
"get_shell_tool_class",
|
||||
"resolve_shell_path",
|
||||
"SUPPORTED_SHELLS",
|
||||
# Core tools
|
||||
"CmdTool",
|
||||
"CmdResult",
|
||||
@@ -144,42 +208,71 @@ __all__ = [
|
||||
def get_shell_tools(
|
||||
permission_manager: PermissionManager,
|
||||
all_tools: dict[str, BaseTool] | None = None,
|
||||
shell_override: str | None = None,
|
||||
) -> list[BaseTool]:
|
||||
"""Create instances of shell tools.
|
||||
|
||||
Only the user's detected/configured shell is included to reduce tool clutter.
|
||||
|
||||
Args:
|
||||
permission_manager: Permission manager for access control
|
||||
all_tools: Dict of all registered tools (for cmd tool invocations)
|
||||
shell_override: Override shell (name like "zsh" or path like "/path/to/fish")
|
||||
|
||||
Returns:
|
||||
List of shell tool instances
|
||||
"""
|
||||
import os
|
||||
|
||||
# Create cmd tool with access to other tools (for tool invocations)
|
||||
cmd = CmdTool(tools=all_tools or {})
|
||||
|
||||
# Create shell wrappers with access to other tools
|
||||
zsh = ZshTool(tools=all_tools or {})
|
||||
bash = BashTool(tools=all_tools or {})
|
||||
|
||||
# Set permission manager for convenience tools
|
||||
npx_tool.permission_manager = permission_manager
|
||||
uvx_tool.permission_manager = permission_manager
|
||||
|
||||
return [
|
||||
cmd, # Primary command execution with DAG support
|
||||
zsh, # Zsh shell wrapper
|
||||
bash, # Bash shell wrapper
|
||||
ps_tool, # Process management
|
||||
# Base tools list
|
||||
tools = [
|
||||
cmd, # Primary command execution with DAG support
|
||||
ps_tool, # Process management
|
||||
npx_tool, # Node packages
|
||||
uvx_tool, # Python packages
|
||||
open_tool, # Open files/URLs
|
||||
open_tool, # Open files/URLs
|
||||
]
|
||||
|
||||
# Check for all-shells mode
|
||||
if os.environ.get("HANZO_MCP_ALL_SHELLS", "").lower() in ("1", "true", "yes"):
|
||||
tools.insert(1, ZshTool(tools=all_tools or {}))
|
||||
tools.insert(2, BashTool(tools=all_tools or {}))
|
||||
tools.insert(3, FishTool(tools=all_tools or {}))
|
||||
tools.insert(4, DashTool(tools=all_tools or {}))
|
||||
return tools
|
||||
|
||||
# Determine which shell to expose
|
||||
if shell_override:
|
||||
# Handle path override (e.g., --shell=/opt/homebrew/bin/fish)
|
||||
if "/" in shell_override:
|
||||
shell_name = os.path.basename(shell_override)
|
||||
os.environ["HANZO_MCP_FORCE_SHELL"] = shell_override
|
||||
else:
|
||||
shell_name = shell_override.lower()
|
||||
else:
|
||||
shell_name, _ = get_cached_active_shell()
|
||||
|
||||
# Create and add only the detected shell tool
|
||||
shell_tool_class = get_shell_tool_class(shell_name)
|
||||
if shell_tool_class:
|
||||
shell_instance = shell_tool_class(tools=all_tools or {})
|
||||
tools.insert(1, shell_instance) # Put shell right after cmd
|
||||
|
||||
return tools
|
||||
|
||||
|
||||
def register_shell_tools(
|
||||
mcp_server: FastMCP,
|
||||
permission_manager: PermissionManager,
|
||||
all_tools: dict[str, BaseTool] | None = None,
|
||||
shell_override: str | None = None,
|
||||
) -> list[BaseTool]:
|
||||
"""Register shell tools with the MCP server.
|
||||
|
||||
@@ -187,11 +280,12 @@ def register_shell_tools(
|
||||
mcp_server: The FastMCP server instance
|
||||
permission_manager: Permission manager for access control
|
||||
all_tools: Dict of all registered tools (for cmd tool invocations)
|
||||
shell_override: Override shell (name like "zsh" or path like "/path/to/fish")
|
||||
|
||||
Returns:
|
||||
List of registered tools
|
||||
"""
|
||||
tools = get_shell_tools(permission_manager, all_tools)
|
||||
tools = get_shell_tools(permission_manager, all_tools, shell_override)
|
||||
ToolRegistry.register_tools(mcp_server, tools)
|
||||
return tools
|
||||
|
||||
@@ -200,9 +294,15 @@ def register_tools(mcp_server: FastMCP, **kwargs) -> list[BaseTool]:
|
||||
"""Register all shell tools with the MCP server.
|
||||
|
||||
This is the standard entry point called by the tool discovery system.
|
||||
|
||||
Supports:
|
||||
- shell_override: Override detected shell (name or path)
|
||||
- permission_manager: Permission manager instance
|
||||
- all_tools: Dict of all registered tools
|
||||
"""
|
||||
from hanzo_tools.core import PermissionManager
|
||||
|
||||
permission_manager = kwargs.get("permission_manager") or PermissionManager()
|
||||
all_tools = kwargs.get("all_tools")
|
||||
return register_shell_tools(mcp_server, permission_manager, all_tools)
|
||||
shell_override = kwargs.get("shell_override") or kwargs.get("shell")
|
||||
return register_shell_tools(mcp_server, permission_manager, all_tools, shell_override)
|
||||
|
||||
@@ -0,0 +1,393 @@
|
||||
"""Shell detection for hanzo-mcp.
|
||||
|
||||
Detects the user's active shell and login shell to expose only
|
||||
the relevant shell tool via MCP. On Mac with Homebrew, this will
|
||||
detect and use /opt/homebrew/bin/zsh if that's what the user has
|
||||
configured.
|
||||
|
||||
Environment variables for override:
|
||||
- HANZO_MCP_SHELL: Force a specific shell (e.g., "zsh", "bash")
|
||||
- HANZO_MCP_FORCE_SHELL: Force a specific shell path (e.g., "/opt/homebrew/bin/zsh")
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import platform
|
||||
import shlex
|
||||
import shutil
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional, Tuple, Dict, List
|
||||
|
||||
|
||||
# Common shells (names you might see from `ps -o comm=`) and typical full paths.
|
||||
KNOWN_SHELL_NAMES = {
|
||||
"zsh", "bash", "fish", "sh", "ksh", "tcsh", "csh", "dash", "nu", "xonsh",
|
||||
"pwsh", "powershell",
|
||||
}
|
||||
KNOWN_SHELL_PATH_BASENAMES = KNOWN_SHELL_NAMES | {"busybox"} # sometimes sh is busybox
|
||||
|
||||
# Shells we support with dedicated tools
|
||||
SUPPORTED_SHELLS = {"zsh", "bash", "fish", "dash"}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ShellInfo:
|
||||
"""Information about detected shells."""
|
||||
# "Login shell" as configured on the user account (what ssh/login uses).
|
||||
login_shell: Optional[str]
|
||||
# The shell that appears to have invoked this process (best-effort).
|
||||
invoking_shell: Optional[str]
|
||||
# $SHELL (often, but not always, matches login shell)
|
||||
env_shell: Optional[str]
|
||||
# Debug evidence for logging / telemetry.
|
||||
evidence: Dict[str, str]
|
||||
|
||||
|
||||
def _run(cmd: List[str]) -> Tuple[int, str, str]:
|
||||
"""Run a command and return (returncode, stdout, stderr)."""
|
||||
try:
|
||||
p = subprocess.run(
|
||||
cmd,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
return p.returncode, p.stdout.strip(), p.stderr.strip()
|
||||
except Exception as e:
|
||||
return 127, "", f"{type(e).__name__}: {e}"
|
||||
|
||||
|
||||
def _which(cmd: str) -> Optional[str]:
|
||||
"""Find command in PATH."""
|
||||
paths = os.environ.get("PATH", "").split(os.pathsep)
|
||||
for d in paths:
|
||||
p = os.path.join(d, cmd)
|
||||
if os.path.isfile(p) and os.access(p, os.X_OK):
|
||||
return p
|
||||
return None
|
||||
|
||||
|
||||
def _parse_passwd_line(line: str) -> Optional[str]:
|
||||
"""Parse passwd format: name:pw:uid:gid:gecos:dir:shell"""
|
||||
parts = line.split(":")
|
||||
if len(parts) >= 7:
|
||||
shell = parts[6].strip()
|
||||
return shell or None
|
||||
return None
|
||||
|
||||
|
||||
def get_login_shell() -> Tuple[Optional[str], Dict[str, str]]:
|
||||
"""
|
||||
Tries multiple sources to determine the *account login shell*.
|
||||
Order of preference:
|
||||
1) Python's pwd database (POSIX)
|
||||
2) macOS: `id -P user` (NSS/passwd view)
|
||||
3) Linux: `getent passwd user`
|
||||
"""
|
||||
evidence: Dict[str, str] = {}
|
||||
user = os.environ.get("USER") or os.environ.get("LOGNAME") or ""
|
||||
|
||||
# POSIX: pwd module uses NSS; on macOS it generally reflects Directory Services.
|
||||
if os.name == "posix":
|
||||
try:
|
||||
import pwd
|
||||
if user:
|
||||
shell = pwd.getpwnam(user).pw_shell
|
||||
evidence["pwd.getpwnam"] = shell
|
||||
if shell:
|
||||
return shell, evidence
|
||||
except Exception as e:
|
||||
evidence["pwd.getpwnam_error"] = f"{type(e).__name__}: {e}"
|
||||
|
||||
system = platform.system().lower()
|
||||
|
||||
# macOS: `id -P user` prints passwd-style record.
|
||||
if system == "darwin" and user:
|
||||
rc, out, err = _run(["id", "-P", user])
|
||||
evidence["id_-P_rc"] = str(rc)
|
||||
if err:
|
||||
evidence["id_-P_err"] = err
|
||||
if rc == 0 and out:
|
||||
evidence["id_-P_out"] = out
|
||||
# id -P output is colon-separated, shell is the last field.
|
||||
shell = out.split(":")[-1].strip() or None
|
||||
if shell:
|
||||
return shell, evidence
|
||||
|
||||
# Linux/BSD: getent is the most robust CLI view into NSS.
|
||||
if user and _which("getent"):
|
||||
rc, out, err = _run(["getent", "passwd", user])
|
||||
evidence["getent_rc"] = str(rc)
|
||||
if err:
|
||||
evidence["getent_err"] = err
|
||||
if rc == 0 and out:
|
||||
evidence["getent_out"] = out
|
||||
shell = _parse_passwd_line(out)
|
||||
if shell:
|
||||
return shell, evidence
|
||||
|
||||
# As a last resort: $SHELL (not authoritative for login shell)
|
||||
env_shell = os.environ.get("SHELL")
|
||||
if env_shell:
|
||||
evidence["fallback_env_SHELL"] = env_shell
|
||||
return env_shell, evidence
|
||||
|
||||
|
||||
def _ps_comm(pid: int) -> Optional[str]:
|
||||
"""Get executable name for a process."""
|
||||
rc, out, _ = _run(["ps", "-p", str(pid), "-o", "comm="])
|
||||
if rc == 0 and out:
|
||||
return out.strip()
|
||||
return None
|
||||
|
||||
|
||||
def _ps_args(pid: int) -> Optional[str]:
|
||||
"""Get full command line for a process."""
|
||||
rc, out, _ = _run(["ps", "-p", str(pid), "-o", "args="])
|
||||
if rc == 0 and out:
|
||||
return out.strip()
|
||||
return None
|
||||
|
||||
|
||||
def get_invoking_shell(max_hops: int = 12) -> Tuple[Optional[str], Dict[str, str]]:
|
||||
"""
|
||||
Best-effort guess of the *shell that invoked this process* by walking parent PIDs.
|
||||
This is useful if hanzo-mcp is launched from a user's interactive shell.
|
||||
If your process is daemonized / launched by a service manager, this may return None.
|
||||
"""
|
||||
evidence: Dict[str, str] = {}
|
||||
if os.name != "posix":
|
||||
return None, {"note": "invoking shell detection is POSIX-only"}
|
||||
|
||||
pid = os.getpid()
|
||||
ppid = os.getppid()
|
||||
evidence["self_pid"] = str(pid)
|
||||
evidence["self_ppid"] = str(ppid)
|
||||
|
||||
# Walk parent chain looking for a known shell.
|
||||
cur = ppid
|
||||
for hop in range(max_hops):
|
||||
comm = _ps_comm(cur) or ""
|
||||
args = _ps_args(cur) or ""
|
||||
evidence[f"hop_{hop}_pid"] = str(cur)
|
||||
if comm:
|
||||
evidence[f"hop_{hop}_comm"] = comm
|
||||
if args:
|
||||
evidence[f"hop_{hop}_args"] = args
|
||||
|
||||
base = os.path.basename(comm).strip()
|
||||
if base in KNOWN_SHELL_PATH_BASENAMES:
|
||||
# Try to return a plausible full path if args contains one.
|
||||
if os.path.isabs(comm):
|
||||
return comm, evidence
|
||||
# Pull first token from args if it looks like a path to shell.
|
||||
try:
|
||||
argv0 = shlex.split(args)[0]
|
||||
except Exception:
|
||||
argv0 = ""
|
||||
if argv0 and (os.path.isabs(argv0) or os.path.basename(argv0) in KNOWN_SHELL_NAMES):
|
||||
return argv0, evidence
|
||||
return comm, evidence
|
||||
|
||||
# Next parent: use `ps -o ppid=` to get parent of parent.
|
||||
rc, out, _ = _run(["ps", "-p", str(cur), "-o", "ppid="])
|
||||
if rc != 0 or not out.strip():
|
||||
break
|
||||
next_ppid = int(out.strip())
|
||||
if next_ppid <= 1 or next_ppid == cur:
|
||||
break
|
||||
cur = next_ppid
|
||||
|
||||
return None, evidence
|
||||
|
||||
|
||||
def normalize_shell(shell: Optional[str]) -> Optional[str]:
|
||||
"""Normalize shell path - return as-is if valid."""
|
||||
if not shell:
|
||||
return None
|
||||
shell = shell.strip()
|
||||
return shell or None
|
||||
|
||||
|
||||
def shell_basename(shell: Optional[str]) -> Optional[str]:
|
||||
"""Get the basename of a shell path (e.g., '/opt/homebrew/bin/zsh' -> 'zsh')."""
|
||||
if not shell:
|
||||
return None
|
||||
return os.path.basename(shell).strip() or None
|
||||
|
||||
|
||||
def choose_default_shell(info: ShellInfo) -> str:
|
||||
"""
|
||||
Policy: prefer invoking shell if detected; else login shell; else $SHELL; else /bin/sh.
|
||||
"""
|
||||
for candidate in (info.invoking_shell, info.login_shell, info.env_shell):
|
||||
c = normalize_shell(candidate)
|
||||
if c:
|
||||
return c
|
||||
# Very conservative fallback
|
||||
return "C:\\Windows\\System32\\cmd.exe" if os.name == "nt" else "/bin/sh"
|
||||
|
||||
|
||||
def detect_shells() -> ShellInfo:
|
||||
"""Detect all shell information."""
|
||||
login_shell, login_ev = get_login_shell()
|
||||
invoking_shell, inv_ev = get_invoking_shell()
|
||||
env_shell = os.environ.get("SHELL")
|
||||
|
||||
evidence: Dict[str, str] = {}
|
||||
evidence.update({f"login.{k}": v for k, v in login_ev.items()})
|
||||
evidence.update({f"invoking.{k}": v for k, v in inv_ev.items()})
|
||||
if env_shell:
|
||||
evidence["env.SHELL"] = env_shell
|
||||
|
||||
return ShellInfo(
|
||||
login_shell=normalize_shell(login_shell),
|
||||
invoking_shell=normalize_shell(invoking_shell),
|
||||
env_shell=normalize_shell(env_shell),
|
||||
evidence=evidence,
|
||||
)
|
||||
|
||||
|
||||
def resolve_shell_path(shell_name: str) -> Optional[str]:
|
||||
"""
|
||||
Resolve a shell name to its full path, preferring Homebrew on macOS.
|
||||
|
||||
Args:
|
||||
shell_name: Shell name (e.g., 'zsh', 'bash')
|
||||
|
||||
Returns:
|
||||
Full path to shell or None if not found
|
||||
"""
|
||||
# Check environment override first
|
||||
force_shell = os.environ.get("HANZO_MCP_FORCE_SHELL")
|
||||
if force_shell:
|
||||
if os.path.isfile(force_shell) and os.access(force_shell, os.X_OK):
|
||||
return force_shell
|
||||
|
||||
# Priority paths - Homebrew first on macOS
|
||||
search_paths = [
|
||||
f"/opt/homebrew/bin/{shell_name}", # Apple Silicon Homebrew
|
||||
f"/usr/local/bin/{shell_name}", # Intel Homebrew
|
||||
f"/bin/{shell_name}", # System
|
||||
f"/usr/bin/{shell_name}", # System alternative
|
||||
]
|
||||
|
||||
for path in search_paths:
|
||||
if os.path.isfile(path) and os.access(path, os.X_OK):
|
||||
return path
|
||||
|
||||
# Fallback to which
|
||||
return shutil.which(shell_name)
|
||||
|
||||
|
||||
def get_active_shell() -> Tuple[str, str]:
|
||||
"""
|
||||
Get the user's active shell name and path.
|
||||
|
||||
Respects environment overrides:
|
||||
- HANZO_MCP_SHELL: Force shell name (e.g., "zsh")
|
||||
- HANZO_MCP_FORCE_SHELL: Force shell path (e.g., "/opt/homebrew/bin/zsh")
|
||||
|
||||
Returns:
|
||||
Tuple of (shell_name, shell_path)
|
||||
e.g., ("zsh", "/opt/homebrew/bin/zsh")
|
||||
"""
|
||||
# Check for explicit override
|
||||
override_shell = os.environ.get("HANZO_MCP_SHELL")
|
||||
if override_shell:
|
||||
override_shell = override_shell.strip().lower()
|
||||
if override_shell in SUPPORTED_SHELLS:
|
||||
path = resolve_shell_path(override_shell)
|
||||
if path:
|
||||
return override_shell, path
|
||||
|
||||
# Check for explicit path override
|
||||
force_path = os.environ.get("HANZO_MCP_FORCE_SHELL")
|
||||
if force_path and os.path.isfile(force_path) and os.access(force_path, os.X_OK):
|
||||
name = shell_basename(force_path)
|
||||
if name and name in SUPPORTED_SHELLS:
|
||||
return name, force_path
|
||||
# If it's an unknown shell, still use it but expose as "shell"
|
||||
return name or "shell", force_path
|
||||
|
||||
# Detect from environment
|
||||
info = detect_shells()
|
||||
chosen_path = choose_default_shell(info)
|
||||
chosen_name = shell_basename(chosen_path)
|
||||
|
||||
# Map to supported shell name
|
||||
if chosen_name in SUPPORTED_SHELLS:
|
||||
# Resolve to best available path (prefer Homebrew)
|
||||
best_path = resolve_shell_path(chosen_name)
|
||||
return chosen_name, best_path or chosen_path
|
||||
|
||||
# Fallback: if detected shell isn't supported, default to zsh
|
||||
for fallback in ["zsh", "bash", "fish", "dash"]:
|
||||
path = resolve_shell_path(fallback)
|
||||
if path:
|
||||
return fallback, path
|
||||
|
||||
# Ultimate fallback
|
||||
return "sh", "/bin/sh"
|
||||
|
||||
|
||||
def get_shell_tool_class(shell_name: str):
|
||||
"""
|
||||
Get the appropriate shell tool class for the given shell name.
|
||||
|
||||
Args:
|
||||
shell_name: Shell name (e.g., 'zsh', 'bash')
|
||||
|
||||
Returns:
|
||||
Shell tool class or None if not supported
|
||||
"""
|
||||
# Import here to avoid circular imports
|
||||
from hanzo_tools.shell.shell_tools import ZshTool, BashTool, FishTool, DashTool
|
||||
|
||||
shell_map = {
|
||||
"zsh": ZshTool,
|
||||
"bash": BashTool,
|
||||
"fish": FishTool,
|
||||
"dash": DashTool,
|
||||
}
|
||||
|
||||
return shell_map.get(shell_name.lower())
|
||||
|
||||
|
||||
# Cache for shell detection (avoid repeated subprocess calls)
|
||||
_cached_shell: Optional[Tuple[str, str]] = None
|
||||
|
||||
|
||||
def get_cached_active_shell() -> Tuple[str, str]:
|
||||
"""Get cached active shell info (detects once per process)."""
|
||||
global _cached_shell
|
||||
if _cached_shell is None:
|
||||
_cached_shell = get_active_shell()
|
||||
return _cached_shell
|
||||
|
||||
|
||||
def clear_shell_cache():
|
||||
"""Clear the shell detection cache (useful for testing)."""
|
||||
global _cached_shell
|
||||
_cached_shell = None
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
info = detect_shells()
|
||||
shell_name, shell_path = get_active_shell()
|
||||
|
||||
print("=== Shell Detection ===")
|
||||
print(f"login_shell : {info.login_shell}")
|
||||
print(f"invoking_shell: {info.invoking_shell}")
|
||||
print(f"env_shell : {info.env_shell}")
|
||||
print(f"")
|
||||
print(f"=== Active Shell ===")
|
||||
print(f"name : {shell_name}")
|
||||
print(f"path : {shell_path}")
|
||||
print(f"")
|
||||
print(f"=== Environment ===")
|
||||
print(f"HANZO_MCP_SHELL : {os.environ.get('HANZO_MCP_SHELL', '(not set)')}")
|
||||
print(f"HANZO_MCP_FORCE_SHELL: {os.environ.get('HANZO_MCP_FORCE_SHELL', '(not set)')}")
|
||||
@@ -4,8 +4,8 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "hanzo-tools-shell"
|
||||
version = "0.5.6"
|
||||
description = "Shell/command execution tools - cmd (unified DAG), zsh, bash, fish, dash, ps, npx, uvx, open, curl, jq, wget"
|
||||
version = "0.6.0"
|
||||
description = "Shell/command execution tools with auto-detection - only exposes your active shell (zsh, bash, fish, dash)"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
license = { text = "MIT" }
|
||||
|
||||
Generated
+1
-1
@@ -440,7 +440,7 @@ wheels = [
|
||||
|
||||
[[package]]
|
||||
name = "hanzo-tools-shell"
|
||||
version = "0.5.6"
|
||||
version = "0.6.0"
|
||||
source = { editable = "." }
|
||||
dependencies = [
|
||||
{ name = "hanzo-async" },
|
||||
|
||||
Reference in New Issue
Block a user