Compare commits

...

10 Commits

Author SHA1 Message Date
Claude4Ξlope 824a66d7c8 Add tmux session management and screen to tmux conversion
Features:
- Add tmux_server.py with full session management capabilities
- Convert TheCoder script from screen to tmux with HOME bind mount fix
- Add enabled flag to MCPServerConfig for selective server startup
- Tmux now default, screen legacy (disabled by default)
- Update documentation and architecture to reflect tmux preference

Session Management:
- create_session, execute, peek, list_sessions, kill_session
- attach_session and share_session with multi-user instructions
- Better multi-user support than screen with native tmux capabilities

Testing:
- Add test_tmux_session.py for comprehensive tmux functionality testing
- Add test_screen_utf8.py for UTF-8 handling
- Add MCP_QUICK_REFERENCE.md for AI handoff documentation

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-06-28 15:28:32 +02:00
Claude4Ξlope edca1d1fe2 Fix gen_apidoc command to properly initialize and discover tools
**Fixed Issues**:
- Proper config file loading from ~/.claude/mcp-browser/config.yaml
- Correct MCPBrowser initialization with config_path parameter
- Fixed server config attribute access using getattr()
- Updated runtime status detection logic
- Reduced initialization wait time for better performance

**Working Features**:
-  Discovers 27 tools from 4 built-in servers (screen, memory, patterns, onboarding)
-  Loads 7 configured external servers from config file
-  Generates comprehensive JSON documentation with runtime status
-  Includes server PIDs and active status for monitoring
-  All regex JSONPath queries working ($.tools[?(@.name =~ /pattern/i)])

The gen_apidoc command now provides complete MCP API documentation
suitable for adding to AI project knowledge bases.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-06-28 13:14:32 +02:00
Claude4Ξlope 1fee302b4d Add regex JSONPath support and gen_apidoc command
**Enhanced JSONPath Support**:
- Manual regex implementation for queries like `$.tools[?(@.name =~ /pattern/i)]`
- Supports case-insensitive searches with /i flag
- Works on both tool names and descriptions
- Fallback to standard JSONPath for non-regex queries

**gen_apidoc Command**:
- New `python setup.py gen_apidoc` command
- Generates comprehensive JSON documentation of MCP API
- Includes all servers, tools, schemas, and capabilities
- AI-optimized format with discovery patterns
- Runtime server status information
- Perfect for adding to project knowledge bases

**API Documentation Features**:
- Complete server inventory with tool counts
- Capability inference from tool names/descriptions
- Discovery pattern examples for common queries
- Sparse mode information for context optimization
- Tool grouping by server with metadata

This enables powerful regex-based tool discovery and provides
comprehensive API documentation for AI consumption.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-06-28 13:04:59 +02:00
Claude4Ξlope c55ffdb59b Add enhanced onboarding examples and Xilope production guide
**Onboarding Enhancements**:
- Added explicit server discovery examples
- Included Claude Code server usage patterns
- Added read_file tool call examples with proper namespacing

**Xilope Production Guide**:
- Complete production environment documentation
- Memory storage integration details (/mnt/data/claude/claude/.mcp-memory/)
- cmem wrapper integration (~/bin/cmem → /usr/local/bin/cmem)
- Bidirectional sync workflows and best practices

**Server Updates**:
- Enhanced onboarding server to serve predefined markdown files
- Priority: predefined files → default.md → fallback instructions

This provides comprehensive guidance for both general MCP Browser usage
and Xilope's specific production environment with cmem integration.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-06-28 12:52:48 +02:00
Claude4Ξlope 63c26975c4 Bump version to 0.2.0 after major enhancements
🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-06-28 12:05:55 +02:00
Claude4Ξlope c8bfa4b2d1 Add enhanced interactive mode and cmem integration
Major enhancements to MCP Browser functionality:

**Interactive Client & Testing**:
- New comprehensive interactive client with readline support
- Tab completion for commands and tools
- Enhanced tool testing with sample argument generation
- Standalone executable script for easy testing

**cmem Integration**:
- Bidirectional sync between memory server and cmem
- Identity-specific memory contexts
- Automatic task, pattern, and decision synchronization
- Graceful degradation when cmem unavailable

**Developer Experience**:
- Updated tool descriptions with clear proxy pattern explanations
- Comprehensive handoff documentation for AI assistant transitions
- 32 new tests covering all enhanced functionality
- All tests passing (46 passed, 3 skipped)

**Context Optimization**:
- Maintained sparse mode efficiency
- Clear meta-tool descriptions with emojis
- Enhanced onboarding guide with practical examples

This update significantly improves both AI assistant workflow continuity
and developer testing capabilities while maintaining the project's core
principle of context efficiency.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-06-28 12:00:52 +02:00
Claude4Ξlope 014b632517 Add daemon management targets to Makefile
- Add 'make run' target that installs and restarts daemon from home directory
- Add 'make restart' target to stop and start daemon without install
- Add 'make stop' target to safely stop running daemon
- Add 'make status' target to check daemon status with PID verification
- Fixes issue where daemon doesn't update after 'make install'
- Ensures daemon runs from .venv/bin/ to avoid stale system-wide version
- Includes process verification with ps command to confirm replacement

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-06-28 00:07:56 +02:00
Claude4Ξlope 8c4ea3a77f Fix license configuration to use SPDX expression
- Replace deprecated License classifier with SPDX license field
- Use "GPL-3.0-or-later" SPDX identifier instead of deprecated classifier
- Removes setuptools deprecation warning

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-06-27 23:51:08 +02:00
Claude4Ξlope 53f672bb76 Add multiuser support to screen server
- Add enable_multiuser tool to enable multiuser mode on sessions
- Add attach_multiuser tool to provide attach instructions
- Add add_user tool for access control list management
- Peek functionality works seamlessly with multiuser sessions
- Create comprehensive test for multiuser functionality
- All existing functionality preserved and tested

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-06-27 23:50:45 +02:00
Claude4Ξlope 140bd2d71b Fix UTF-8 encoding issue in screen peek
- Read screen hardcopy output as binary first
- Decode with UTF-8 using 'replace' error handling
- Fallback to latin-1 encoding if needed
- Strip ANSI escape sequences from output
- Handles mixed encodings and binary data gracefully

This fixes crashes when screen sessions contain non-UTF-8 bytes
or terminal control sequences.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-06-27 23:20:53 +02:00
28 changed files with 3677 additions and 66 deletions

99
.tags
View File

@ -18,6 +18,7 @@ DEFAULT_CONFIGS mcp_browser/default_configs.py /^DEFAULT_CONFIGS = {$/;" v
Decision build/lib/mcp_servers/memory/memory_server.py /^class Decision:$/;" c
Decision mcp_servers/memory/memory_server.py /^class Decision:$/;" c
GenerateAIDocs setup.py /^class GenerateAIDocs(Command):$/;" c
InteractiveMCPClient mcp_browser/interactive_client.py /^class InteractiveMCPClient:$/;" c
JSONRPCTestClient tests/test_integration.py /^class JSONRPCTestClient:$/;" c
JsonRpcBuffer build/lib/mcp_browser/buffer.py /^class JsonRpcBuffer:$/;" c
JsonRpcBuffer mcp_browser/buffer.py /^class JsonRpcBuffer:$/;" c
@ -55,7 +56,10 @@ TRACE build/lib/mcp_browser/logging_config.py /^TRACE = 5$/;" v
TRACE mcp_browser/logging_config.py /^TRACE = 5$/;" v
Task build/lib/mcp_servers/memory/memory_server.py /^class Task:$/;" c
Task mcp_servers/memory/memory_server.py /^class Task:$/;" c
TestCmemIntegration tests/test_cmem_integration.py /^class TestCmemIntegration:$/;" c
TestCommand setup.py /^class TestCommand(Command):$/;" c
TestInteractiveMCPClient tests/test_interactive_client.py /^class TestInteractiveMCPClient:$/;" c
TestInteractiveMCPClientIntegration tests/test_interactive_client.py /^class TestInteractiveMCPClientIntegration:$/;" c
TestMCPBrowser tests/test_basic.py /^class TestMCPBrowser:$/;" c
TestMessageFilter tests/test_basic.py /^class TestMessageFilter:$/;" c
TestToolRegistry tests/test_basic.py /^class TestToolRegistry:$/;" c
@ -103,12 +107,13 @@ __init__ mcp_browser/daemon.py /^ def __init__(self, socket_path: Path):$/;"
__init__ mcp_browser/default_configs.py /^ def __init__(self, config_dir: Optional[Path] = None):$/;" m class:ConfigManager
__init__ mcp_browser/filter.py /^ def __init__(self, registry: ToolRegistry, server_callback: Callable):$/;" m class:VirtualToolHandler
__init__ mcp_browser/filter.py /^ def __init__(self, registry: ToolRegistry, sparse_mode: bool = True):$/;" m class:MessageFilter
__init__ mcp_browser/interactive_client.py /^ def __init__(self, server_name: Optional[str] = None, use_daemon: bool = True):$/;" m class:InteractiveMCPClient
__init__ mcp_browser/multi_server.py /^ def __init__(self, logger=None):$/;" m class:MultiServerManager
__init__ mcp_browser/proxy.py /^ def __init__(self, config_path: Optional[Path] = None, server_name: Optional[str] = None,$/;" m class:MCPBrowser
__init__ mcp_browser/registry.py /^ def __init__(self):$/;" m class:ToolRegistry
__init__ mcp_browser/server.py /^ def __init__(self, config: MCPServerConfig, logger: Optional[logging.Logger] = None):$/;" m class:MCPServer
__init__ mcp_servers/base.py /^ def __init__(self, name: str, version: str = "1.0.0"):$/;" m class:BaseMCPServer
__init__ mcp_servers/memory/memory_server.py /^ def __init__(self):$/;" m class:MemoryServer
__init__ mcp_servers/memory/memory_server.py /^ def __init__(self, identity: str = "default"):$/;" m class:MemoryServer
__init__ mcp_servers/onboarding/onboarding_server.py /^ def __init__(self):$/;" m class:OnboardingServer
__init__ mcp_servers/pattern_manager/pattern_server.py /^ def __init__(self):$/;" m class:PatternServer
__init__ mcp_servers/screen/screen_server.py /^ def __init__(self):$/;" m class:ScreenServer
@ -119,20 +124,35 @@ __version__ build/lib/mcp_browser/__init__.py /^__version__ = "0.1.0"$/;" v
__version__ mcp_browser/__init__.py /^__version__ = "0.1.0"$/;" v
_add_pattern build/lib/mcp_servers/pattern_manager/pattern_server.py /^ async def _add_pattern(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:PatternServer
_add_pattern mcp_servers/pattern_manager/pattern_server.py /^ async def _add_pattern(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:PatternServer
_add_user build/lib/mcp_servers/screen/screen_server.py /^ async def _add_user(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:ScreenServer
_add_user mcp_servers/screen/screen_server.py /^ async def _add_user(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:ScreenServer
_attach_multiuser build/lib/mcp_servers/screen/screen_server.py /^ async def _attach_multiuser(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:ScreenServer
_attach_multiuser mcp_servers/screen/screen_server.py /^ async def _attach_multiuser(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:ScreenServer
_calculate_checksum build/lib/mcp_browser/default_configs.py /^ def _calculate_checksum(self, filepath: Path) -> str:$/;" m class:ConfigManager
_calculate_checksum mcp_browser/default_configs.py /^ def _calculate_checksum(self, filepath: Path) -> str:$/;" m class:ConfigManager
_call_mcp mcp_browser/interactive_client.py /^ async def _call_mcp(self, request: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:InteractiveMCPClient
_call_tool mcp_browser/interactive_client.py /^ async def _call_tool(self, args: List[str]):$/;" f
_call_tool_direct mcp_browser/interactive_client.py /^ async def _call_tool_direct(self, tool_name: str, args: List[str]):$/;" f
_completer mcp_browser/interactive_client.py /^ def _completer(self, text: str, state: int) -> Optional[str]:$/;" m class:InteractiveMCPClient
_create_cmem_bridges mcp_servers/memory/memory_server.py /^ def _create_cmem_bridges(self, identity_dir: Path, session_dir: Path):$/;" m class:MemoryServer
_create_session build/lib/mcp_servers/screen/screen_server.py /^ async def _create_session(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:ScreenServer
_create_session mcp_servers/screen/screen_server.py /^ async def _create_session(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:ScreenServer
_decision_add build/lib/mcp_servers/memory/memory_server.py /^ async def _decision_add(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:MemoryServer
_decision_add mcp_servers/memory/memory_server.py /^ async def _decision_add(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:MemoryServer
_delete_onboarding build/lib/mcp_servers/onboarding/onboarding_server.py /^ async def _delete_onboarding(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:OnboardingServer
_delete_onboarding mcp_servers/onboarding/onboarding_server.py /^ async def _delete_onboarding(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:OnboardingServer
_discover_tools mcp_browser/interactive_client.py /^ async def _discover_tools(self, args: List[str]):$/;" f
_display_result mcp_browser/interactive_client.py /^ def _display_result(self, result: Any):$/;" f
_enable_multiuser build/lib/mcp_servers/screen/screen_server.py /^ async def _enable_multiuser(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:ScreenServer
_enable_multiuser mcp_servers/screen/screen_server.py /^ async def _enable_multiuser(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:ScreenServer
_ensure_config_file build/lib/mcp_browser/default_configs.py /^ def _ensure_config_file(self, filename: str, config_data: Dict[str, str]) -> None:$/;" m class:ConfigManager
_ensure_config_file mcp_browser/default_configs.py /^ def _ensure_config_file(self, filename: str, config_data: Dict[str, str]) -> None:$/;" m class:ConfigManager
_execute_command build/lib/mcp_servers/screen/screen_server.py /^ async def _execute_command(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:ScreenServer
_execute_command mcp_browser/interactive_client.py /^ async def _execute_command(self, line: str):$/;" m class:InteractiveMCPClient
_execute_command mcp_servers/screen/screen_server.py /^ async def _execute_command(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:ScreenServer
_execute_pattern build/lib/mcp_servers/pattern_manager/pattern_server.py /^ async def _execute_pattern(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:PatternServer
_execute_pattern mcp_servers/pattern_manager/pattern_server.py /^ async def _execute_pattern(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:PatternServer
_execute_tool_call mcp_browser/interactive_client.py /^ async def _execute_tool_call(self, tool_name: str, arguments: Dict[str, Any]):$/;" f
_export_onboarding build/lib/mcp_servers/onboarding/onboarding_server.py /^ async def _export_onboarding(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:OnboardingServer
_export_onboarding mcp_servers/onboarding/onboarding_server.py /^ async def _export_onboarding(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:OnboardingServer
_filter_tools_response build/lib/mcp_browser/filter.py /^ def _filter_tools_response(self, message: dict) -> dict:$/;" m class:MessageFilter
@ -141,6 +161,7 @@ _format_onboarding build/lib/mcp_servers/onboarding/onboarding_server.py /^ d
_format_onboarding mcp_servers/onboarding/onboarding_server.py /^ def _format_onboarding(self, identity: str, data: Dict[str, Any]) -> str:$/;" m class:OnboardingServer
_forward_to_server build/lib/mcp_browser/proxy.py /^ async def _forward_to_server(self, request: dict) -> dict:$/;" m class:MCPBrowser
_forward_to_server mcp_browser/proxy.py /^ async def _forward_to_server(self, request: dict) -> dict:$/;" m class:MCPBrowser
_generate_sample_args mcp_browser/interactive_client.py /^ def _generate_sample_args(self, schema: Dict[str, Any]) -> Dict[str, Any]:$/;" f
_get_builtin_servers build/lib/mcp_browser/multi_server.py /^ def _get_builtin_servers(self) -> Dict[str, MCPServerConfig]:$/;" m class:MultiServerManager
_get_builtin_servers mcp_browser/multi_server.py /^ def _get_builtin_servers(self) -> Dict[str, MCPServerConfig]:$/;" m class:MultiServerManager
_handle_call build/lib/mcp_browser/filter.py /^ async def _handle_call(self, message: dict) -> dict:$/;" m class:VirtualToolHandler
@ -169,12 +190,14 @@ _list_patterns build/lib/mcp_servers/pattern_manager/pattern_server.py /^ asy
_list_patterns mcp_servers/pattern_manager/pattern_server.py /^ async def _list_patterns(self) -> Dict[str, Any]:$/;" m class:PatternServer
_list_sessions build/lib/mcp_servers/screen/screen_server.py /^ async def _list_sessions(self) -> Dict[str, Any]:$/;" m class:ScreenServer
_list_sessions mcp_servers/screen/screen_server.py /^ async def _list_sessions(self) -> Dict[str, Any]:$/;" m class:ScreenServer
_list_tools mcp_browser/interactive_client.py /^ async def _list_tools(self, args: List[str]):$/;" f
_load_json build/lib/mcp_servers/memory/memory_server.py /^ def _load_json(self, filename: str, default: Any) -> Any:$/;" m class:MemoryServer
_load_json mcp_servers/memory/memory_server.py /^ def _load_json(self, filename: str, default: Any) -> Any:$/;" m class:MemoryServer
_load_memory build/lib/mcp_servers/memory/memory_server.py /^ def _load_memory(self):$/;" m class:MemoryServer
_load_memory mcp_servers/memory/memory_server.py /^ def _load_memory(self):$/;" m class:MemoryServer
_load_patterns build/lib/mcp_servers/pattern_manager/pattern_server.py /^ def _load_patterns(self) -> Dict[str, Dict[str, Any]]:$/;" m class:PatternServer
_load_patterns mcp_servers/pattern_manager/pattern_server.py /^ def _load_patterns(self) -> Dict[str, Dict[str, Any]]:$/;" m class:PatternServer
_manage_onboarding mcp_browser/interactive_client.py /^ async def _manage_onboarding(self, args: List[str]):$/;" f
_mark_offline build/lib/mcp_browser/server.py /^ def _mark_offline(self):$/;" m class:MCPServer
_mark_offline mcp_browser/server.py /^ def _mark_offline(self):$/;" m class:MCPServer
_matches_pattern build/lib/mcp_servers/pattern_manager/pattern_server.py /^ def _matches_pattern(self, text: str, trigger: List[str]) -> bool:$/;" m class:PatternServer
@ -201,6 +224,7 @@ _read_stderr build/lib/mcp_browser/server.py /^ async def _read_stderr(self):
_read_stderr mcp_browser/server.py /^ async def _read_stderr(self):$/;" m class:MCPServer
_read_stdout build/lib/mcp_browser/server.py /^ async def _read_stdout(self):$/;" m class:MCPServer
_read_stdout mcp_browser/server.py /^ async def _read_stdout(self):$/;" m class:MCPServer
_refresh_tools mcp_browser/interactive_client.py /^ async def _refresh_tools(self):$/;" m class:InteractiveMCPClient
_register_tools build/lib/mcp_servers/memory/memory_server.py /^ def _register_tools(self):$/;" m class:MemoryServer
_register_tools build/lib/mcp_servers/onboarding/onboarding_server.py /^ def _register_tools(self):$/;" m class:OnboardingServer
_register_tools build/lib/mcp_servers/pattern_manager/pattern_server.py /^ def _register_tools(self):$/;" m class:PatternServer
@ -219,8 +243,17 @@ _save_json build/lib/mcp_servers/memory/memory_server.py /^ def _save_json(se
_save_json mcp_servers/memory/memory_server.py /^ def _save_json(self, filename: str, data: Any):$/;" m class:MemoryServer
_save_patterns build/lib/mcp_servers/pattern_manager/pattern_server.py /^ def _save_patterns(self):$/;" m class:PatternServer
_save_patterns mcp_servers/pattern_manager/pattern_server.py /^ def _save_patterns(self):$/;" m class:PatternServer
_setup_cmem_integration mcp_servers/memory/memory_server.py /^ def _setup_cmem_integration(self) -> bool:$/;" m class:MemoryServer
_setup_readline mcp_browser/interactive_client.py /^ def _setup_readline(self):$/;" m class:InteractiveMCPClient
_show_help mcp_browser/interactive_client.py /^ def _show_help(self):$/;" m class:InteractiveMCPClient
_show_status mcp_browser/interactive_client.py /^ async def _show_status(self):$/;" f
_signal_handler build/lib/mcp_browser/daemon.py /^ def _signal_handler(self, signum, frame):$/;" m class:MCPBrowserDaemon
_signal_handler mcp_browser/daemon.py /^ def _signal_handler(self, signum, frame):$/;" m class:MCPBrowserDaemon
_start_config_watcher build/lib/mcp_browser/proxy.py /^ async def _start_config_watcher(self):$/;" m class:MCPBrowser
_start_config_watcher mcp_browser/proxy.py /^ async def _start_config_watcher(self):$/;" m class:MCPBrowser
_sync_decision_to_cmem mcp_servers/memory/memory_server.py /^ async def _sync_decision_to_cmem(self, decision: Decision):$/;" m class:MemoryServer
_sync_pattern_to_cmem mcp_servers/memory/memory_server.py /^ async def _sync_pattern_to_cmem(self, pattern: Pattern, action: str):$/;" m class:MemoryServer
_sync_task_to_cmem mcp_servers/memory/memory_server.py /^ async def _sync_task_to_cmem(self, task: Task, action: str):$/;" m class:MemoryServer
_task_add build/lib/mcp_servers/memory/memory_server.py /^ async def _task_add(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:MemoryServer
_task_add mcp_servers/memory/memory_server.py /^ async def _task_add(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:MemoryServer
_task_list build/lib/mcp_servers/memory/memory_server.py /^ async def _task_list(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:MemoryServer
@ -229,6 +262,9 @@ _task_update build/lib/mcp_servers/memory/memory_server.py /^ async def _task
_task_update mcp_servers/memory/memory_server.py /^ async def _task_update(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:MemoryServer
_test_pattern build/lib/mcp_servers/pattern_manager/pattern_server.py /^ async def _test_pattern(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:PatternServer
_test_pattern mcp_servers/pattern_manager/pattern_server.py /^ async def _test_pattern(self, args: Dict[str, Any]) -> Dict[str, Any]:$/;" m class:PatternServer
_test_tool mcp_browser/interactive_client.py /^ async def _test_tool(self, args: List[str]):$/;" f
_update_server_configs build/lib/mcp_browser/proxy.py /^ def _update_server_configs(self):$/;" m class:MCPBrowser
_update_server_configs mcp_browser/proxy.py /^ def _update_server_configs(self):$/;" m class:MCPBrowser
add_message_handler build/lib/mcp_browser/server.py /^ def add_message_handler(self, handler: Callable[[dict], None]):$/;" m class:MCPServer
add_message_handler mcp_browser/server.py /^ def add_message_handler(self, handler: Callable[[dict], None]):$/;" m class:MCPServer
add_server build/lib/mcp_browser/multi_server.py /^ async def add_server(self, name: str, config: MCPServerConfig):$/;" m class:MultiServerManager
@ -254,6 +290,7 @@ call mcp_browser/proxy.py /^ async def call(self, jsonrpc_object: Dict[str, A
call_tool build/lib/mcp_browser/__main__.py /^ call_tool = subparsers.add_parser("tools-call", help="Call a tool")$/;" v
call_tool mcp_browser/__main__.py /^ call_tool = subparsers.add_parser("tools-call", help="Call a tool")$/;" v
classifiers setup.py /^ classifiers=[$/;" v
cleanup mcp_browser/interactive_client.py /^ async def cleanup(self):$/;" f
clear build/lib/mcp_browser/buffer.py /^ def clear(self):$/;" m class:JsonRpcBuffer
clear mcp_browser/buffer.py /^ def clear(self):$/;" m class:JsonRpcBuffer
close build/lib/mcp_browser/daemon.py /^ async def close(self):$/;" m class:MCPBrowserClient
@ -264,8 +301,6 @@ close mcp_browser/proxy.py /^ async def close(self):$/;" m class:MCPBrowser
cmdclass setup.py /^ cmdclass={$/;" v
completion build/lib/mcp_browser/__main__.py /^ completion = subparsers.add_parser("completion", help="Get completion")$/;" v
completion mcp_browser/__main__.py /^ completion = subparsers.add_parser("completion", help="Get completion")$/;" v
config build/lib/mcp_browser/__main__.py /^ config = loader.load()$/;" v
config mcp_browser/__main__.py /^ config = loader.load()$/;" v
config_path build/lib/mcp_browser/__main__.py /^ config_path=config_path,$/;" v
config_path build/lib/mcp_browser/__main__.py /^ config_path = Path(args.config) if args.config else None$/;" v
config_path mcp_browser/__main__.py /^ config_path=config_path,$/;" v
@ -377,6 +412,7 @@ help mcp_browser/__main__.py /^ help="Test connection to s
include_package_data setup.py /^ include_package_data=True,$/;" v
initialize build/lib/mcp_browser/proxy.py /^ async def initialize(self):$/;" m class:MCPBrowser
initialize examples/ai_optimized.py /^ async def initialize(self):$/;" m class:AIAssistant
initialize mcp_browser/interactive_client.py /^ async def initialize(self):$/;" m class:InteractiveMCPClient
initialize mcp_browser/proxy.py /^ async def initialize(self):$/;" m class:MCPBrowser
initialize_options setup.py /^ def initialize_options(self):$/;" m class:GenerateAIDocs
initialize_options setup.py /^ def initialize_options(self):$/;" m class:TestCommand
@ -394,18 +430,14 @@ jsonrpc mcp_browser/__main__.py /^ jsonrpc = subparsers.add_parser("jsonrpc",
keywords setup.py /^ keywords="mcp model-context-protocol ai llm tools json-rpc",$/;" v
kill_daemon_with_children build/lib/mcp_browser/daemon.py /^def kill_daemon_with_children(socket_path: Path) -> bool:$/;" f
kill_daemon_with_children mcp_browser/daemon.py /^def kill_daemon_with_children(socket_path: Path) -> bool:$/;" f
list_prompts build/lib/mcp_browser/__main__.py /^ list_prompts = subparsers.add_parser("prompts-list", help="List available prompts")$/;" v
list_prompts mcp_browser/__main__.py /^ list_prompts = subparsers.add_parser("prompts-list", help="List available prompts")$/;" v
list_resources build/lib/mcp_browser/__main__.py /^ list_resources = subparsers.add_parser("resources-list", help="List available resources")$/;" v
list_resources mcp_browser/__main__.py /^ list_resources = subparsers.add_parser("resources-list", help="List available resources")$/;" v
list_tools build/lib/mcp_browser/__main__.py /^ list_tools = subparsers.add_parser("tools-list", help="List available tools")$/;" v
list_tools mcp_browser/__main__.py /^ list_tools = subparsers.add_parser("tools-list", help="List available tools")$/;" v
license setup.py /^ license="GPL-3.0-or-later",$/;" v
load build/lib/mcp_browser/config.py /^ def load(self) -> MCPBrowserConfig:$/;" m class:ConfigLoader
load mcp_browser/config.py /^ def load(self) -> MCPBrowserConfig:$/;" m class:ConfigLoader
load_config build/lib/mcp_browser/default_configs.py /^ def load_config(self) -> dict:$/;" m class:ConfigManager
load_config mcp_browser/default_configs.py /^ def load_config(self) -> dict:$/;" m class:ConfigManager
loader build/lib/mcp_browser/__main__.py /^ loader = ConfigLoader()$/;" v
loader mcp_browser/__main__.py /^ loader = ConfigLoader()$/;" v
log_test test_mcp_protocol.py /^def log_test(msg):$/;" f
long_description setup.py /^ long_description = fh.read()$/;" v
long_description setup.py /^ long_description=long_description,$/;" v
long_description_content_type setup.py /^ long_description_content_type="text\/markdown",$/;" v
@ -417,10 +449,13 @@ main examples/basic_usage.py /^async def main():$/;" f
main mcp_browser/__main__.py /^def main():$/;" f
main mcp_browser/client_main.py /^def main():$/;" f
main mcp_browser/daemon_main.py /^def main():$/;" f
main mcp_browser/interactive_client.py /^async def main():$/;" f
main test_mcp_protocol.py /^async def main():$/;" f
main tests/test_browser_functionality.py /^def main():$/;" f
main tests/test_integration.py /^async def main():$/;" f
mark_handled build/lib/mcp_browser/filter.py /^ def mark_handled(self, request_id: Union[str, int]):$/;" m class:MessageFilter
mark_handled mcp_browser/filter.py /^ def mark_handled(self, request_id: Union[str, int]):$/;" m class:MessageFilter
mock_call tests/test_interactive_client.py /^ def mock_call(request):$/;" f function:TestInteractiveMCPClientIntegration.test_full_workflow_mock
name setup.py /^ name="mcp-browser",$/;" v
package_data setup.py /^ package_data={$/;" v
packages setup.py /^ packages=find_packages(include=['mcp_browser*', 'mcp_servers*']),$/;" v
@ -439,6 +474,7 @@ route_tool_call mcp_browser/multi_server.py /^ async def route_tool_call(self
run build/lib/mcp_browser/client_main.py /^ async def run():$/;" f function:main
run build/lib/mcp_servers/base.py /^ async def run(self):$/;" m class:BaseMCPServer
run mcp_browser/client_main.py /^ async def run():$/;" f function:main
run mcp_browser/interactive_client.py /^ async def run(self):$/;" m class:InteractiveMCPClient
run mcp_servers/base.py /^ async def run(self):$/;" m class:BaseMCPServer
run setup.py /^ def run(self):$/;" m class:GenerateAIDocs
run setup.py /^ def run(self):$/;" m class:TestCommand
@ -477,6 +513,8 @@ set_onboarding_text build/lib/mcp_browser/default_configs.py /^ def set_onboa
set_onboarding_text mcp_browser/default_configs.py /^ def set_onboarding_text(self, text: str, identity: Optional[str] = None) -> None:$/;" m class:ConfigManager
setup_logging build/lib/mcp_browser/logging_config.py /^def setup_logging(debug: bool = False, log_file: Optional[Path] = None, $/;" f
setup_logging mcp_browser/logging_config.py /^def setup_logging(debug: bool = False, log_file: Optional[Path] = None, $/;" f
setup_method tests/test_cmem_integration.py /^ def setup_method(self):$/;" m class:TestCmemIntegration
setup_method tests/test_interactive_client.py /^ def setup_method(self):$/;" m class:TestInteractiveMCPClient
show_available_servers build/lib/mcp_browser/__main__.py /^def show_available_servers(config_path: Optional[str] = None):$/;" f
show_available_servers mcp_browser/__main__.py /^def show_available_servers(config_path: Optional[str] = None):$/;" f
show_configuration build/lib/mcp_browser/__main__.py /^def show_configuration(config_path: Optional[str] = None):$/;" f
@ -511,20 +549,61 @@ test_basic_flow tests/test_integration.py /^async def test_basic_flow():$/;" f
test_brave_search_integration tests/test_brave_search.py /^async def test_brave_search_integration():$/;" f
test_browser_without_servers tests/test_basic.py /^ async def test_browser_without_servers(self):$/;" m class:TestMCPBrowser
test_browser_without_servers tests/test_browser_functionality.py /^async def test_browser_without_servers():$/;" f
test_call_mcp_browser tests/test_interactive_client.py /^ async def test_call_mcp_browser(self):$/;" m class:TestInteractiveMCPClient
test_call_mcp_client tests/test_interactive_client.py /^ async def test_call_mcp_client(self):$/;" m class:TestInteractiveMCPClient
test_claude_connection tests/test_claude_connection.py /^async def test_claude_connection():$/;" f
test_completer_commands tests/test_interactive_client.py /^ def test_completer_commands(self):$/;" m class:TestInteractiveMCPClient
test_completer_tools tests/test_interactive_client.py /^ def test_completer_tools(self):$/;" m class:TestInteractiveMCPClient
test_create_cmem_bridges tests/test_cmem_integration.py /^ def test_create_cmem_bridges(self):$/;" m class:TestCmemIntegration
test_daemon_initialization test_mcp_protocol.py /^async def test_daemon_initialization():$/;" f
test_direct_initialization test_mcp_protocol.py /^async def test_direct_initialization():$/;" f
test_discover_jsonpath tests/test_basic.py /^ def test_discover_jsonpath(self):$/;" m class:TestToolRegistry
test_discovery test_discovery.py /^async def test_discovery():$/;" f
test_display_result_image_content tests/test_interactive_client.py /^ def test_display_result_image_content(self):$/;" m class:TestInteractiveMCPClient
test_display_result_raw_data tests/test_interactive_client.py /^ def test_display_result_raw_data(self):$/;" m class:TestInteractiveMCPClient
test_display_result_text_content tests/test_interactive_client.py /^ def test_display_result_text_content(self):$/;" m class:TestInteractiveMCPClient
test_double_handshake_issue test_mcp_protocol.py /^async def test_double_handshake_issue():$/;" f
test_duplicate_error_filtering tests/test_basic.py /^ def test_duplicate_error_filtering(self):$/;" m class:TestMessageFilter
test_error_handling tests/test_integration.py /^async def test_error_handling():$/;" f
test_execute_command_help tests/test_interactive_client.py /^ async def test_execute_command_help(self):$/;" m class:TestInteractiveMCPClient
test_execute_command_list tests/test_interactive_client.py /^ async def test_execute_command_list(self):$/;" m class:TestInteractiveMCPClient
test_execute_command_refresh tests/test_interactive_client.py /^ async def test_execute_command_refresh(self):$/;" m class:TestInteractiveMCPClient
test_execute_command_unknown_tool tests/test_interactive_client.py /^ async def test_execute_command_unknown_tool(self):$/;" m class:TestInteractiveMCPClient
test_execute_tool_call tests/test_interactive_client.py /^ async def test_execute_tool_call(self):$/;" m class:TestInteractiveMCPClient
test_execute_tool_call_error tests/test_interactive_client.py /^ async def test_execute_tool_call_error(self):$/;" m class:TestInteractiveMCPClient
test_full_workflow_mock tests/test_interactive_client.py /^ async def test_full_workflow_mock(self):$/;" m class:TestInteractiveMCPClientIntegration
test_generate_sample_args_examples tests/test_interactive_client.py /^ def test_generate_sample_args_examples(self):$/;" m class:TestInteractiveMCPClient
test_generate_sample_args_string tests/test_interactive_client.py /^ def test_generate_sample_args_string(self):$/;" m class:TestInteractiveMCPClient
test_generate_sample_args_types tests/test_interactive_client.py /^ def test_generate_sample_args_types(self):$/;" m class:TestInteractiveMCPClient
test_initialization tests/test_interactive_client.py /^ def test_initialization(self):$/;" m class:TestInteractiveMCPClient
test_memory_server_initialization_custom_identity tests/test_cmem_integration.py /^ def test_memory_server_initialization_custom_identity(self):$/;" m class:TestCmemIntegration
test_memory_server_initialization_default tests/test_cmem_integration.py /^ def test_memory_server_initialization_default(self):$/;" m class:TestCmemIntegration
test_message_filter tests/test_simple.py /^def test_message_filter():$/;" f
test_onboarding tests/test_onboarding.py /^async def test_onboarding():$/;" f
test_refresh_tools tests/test_interactive_client.py /^ async def test_refresh_tools(self):$/;" m class:TestInteractiveMCPClient
test_screen_multiuser test_screen_multiuser.py /^async def test_screen_multiuser():$/;" f
test_screen_utf8 test_screen_utf8.py /^async def test_screen_utf8():$/;" f
test_server_connection build/lib/mcp_browser/__main__.py /^async def test_server_connection(browser: MCPBrowser, server_name: Optional[str] = None):$/;" f
test_server_connection mcp_browser/__main__.py /^async def test_server_connection(browser: MCPBrowser, server_name: Optional[str] = None):$/;" f
test_server_mode_initialization test_mcp_protocol.py /^async def test_server_mode_initialization():$/;" f
test_server_mode_timeout test_claude_desktop_flow.py /^async def test_server_mode_timeout():$/;" f
test_setup_cmem_integration_available tests/test_cmem_integration.py /^ def test_setup_cmem_integration_available(self):$/;" m class:TestCmemIntegration
test_setup_cmem_integration_no_cmem tests/test_cmem_integration.py /^ def test_setup_cmem_integration_no_cmem(self):$/;" m class:TestCmemIntegration
test_sparse_mode tests/test_simple.py /^def test_sparse_mode():$/;" f
test_sparse_mode_filtering tests/test_basic.py /^ def test_sparse_mode_filtering(self):$/;" m class:TestMessageFilter
test_sparse_tools tests/test_basic.py /^ def test_sparse_tools(self):$/;" m class:TestToolRegistry
test_sync_decision_to_cmem tests/test_cmem_integration.py /^ async def test_sync_decision_to_cmem(self):$/;" m class:TestCmemIntegration
test_sync_error_handling tests/test_cmem_integration.py /^ async def test_sync_error_handling(self):$/;" m class:TestCmemIntegration
test_sync_pattern_to_cmem tests/test_cmem_integration.py /^ async def test_sync_pattern_to_cmem(self):$/;" m class:TestCmemIntegration
test_sync_task_to_cmem_add tests/test_cmem_integration.py /^ async def test_sync_task_to_cmem_add(self):$/;" m class:TestCmemIntegration
test_sync_task_to_cmem_complete tests/test_cmem_integration.py /^ async def test_sync_task_to_cmem_complete(self):$/;" m class:TestCmemIntegration
test_sync_with_integration_disabled tests/test_cmem_integration.py /^ async def test_sync_with_integration_disabled(self):$/;" m class:TestCmemIntegration
test_task_add_with_sync tests/test_cmem_integration.py /^ async def test_task_add_with_sync(self):$/;" m class:TestCmemIntegration
test_task_update_completion_with_sync tests/test_cmem_integration.py /^ async def test_task_update_completion_with_sync(self):$/;" m class:TestCmemIntegration
test_tool_registry tests/test_simple.py /^def test_tool_registry():$/;" f
test_update_tools tests/test_basic.py /^ def test_update_tools(self):$/;" m class:TestToolRegistry
test_virtual_tool_handling tests/test_basic.py /^ async def test_virtual_tool_handling(self):$/;" m class:TestMCPBrowser
test_with_logging test_claude_desktop_flow.py /^async def test_with_logging():$/;" f
to_json build/lib/mcp_browser/registry.py /^ def to_json(self) -> str:$/;" m class:ToolRegistry
to_json mcp_browser/registry.py /^ def to_json(self) -> str:$/;" m class:ToolRegistry
trace build/lib/mcp_browser/logging_config.py /^def trace(self, message, *args, **kwargs):$/;" f
@ -539,3 +618,5 @@ user_options setup.py /^ user_options = []$/;" v class:TestCommand
version build/lib/mcp_browser/__main__.py /^ version=f"%(prog)s {__version__}",$/;" v
version mcp_browser/__main__.py /^ version=f"%(prog)s {__version__}",$/;" v
version setup.py /^ version="0.1.0",$/;" v
watch_config build/lib/mcp_browser/proxy.py /^ async def watch_config():$/;" f function:MCPBrowser._start_config_watcher
watch_config mcp_browser/proxy.py /^ async def watch_config():$/;" f function:MCPBrowser._start_config_watcher

View File

@ -40,8 +40,9 @@ python setup.py aidocs
1. **Virtual Tools**: `mcp_discover`, `mcp_call`, and `onboarding` exist only in the browser layer
2. **Tool Namespacing**: Format is `server::tool` or `mcp__namespace__tool`
3. **Multi-Server**: Built-in servers (screen, memory, patterns, onboarding) start automatically
3. **Multi-Server**: Built-in servers (tmux, memory, patterns, onboarding) start automatically
4. **Identity-Aware**: Onboarding tool accepts identity parameter for context-specific instructions
5. **Session Management**: Tmux preferred over screen for better multi-user support
### Architecture Overview
@ -55,7 +56,7 @@ mcp_browser/
mcp_servers/
├── base.py # Base class for Python MCP servers
├── screen/ # GNU screen session management
├── screen/ # Session management (tmux preferred, screen legacy)
├── memory/ # Persistent memory and tasks
├── patterns/ # Auto-response patterns
└── onboarding/ # Identity-aware onboarding

267
HANDOFF_INSTRUCTIONS.md Normal file
View File

@ -0,0 +1,267 @@
# Claude Memory (cmem) Handoff System Guide
## Overview
The `cmem` (Claude Memory) handoff system enables seamless AI assistant transitions by providing persistent memory and context across sessions. This is critical for long-running projects where different AI instances need to understand previous work and continue development effectively.
## Core Concepts
### 1. Session-Based Memory
- **Sessions**: Each AI work session is tracked with timestamps and outcomes
- **Auto-rotation**: Sessions automatically rotate after 4 hours to maintain context freshness
- **Session Names**: Descriptive names like "Morning Development", "Late Night Development"
### 2. Structured Knowledge Types
- **Tasks**: Open, in-progress, and completed work items with priorities and assignees
- **Decisions**: Important choices with reasoning and alternatives considered
- **Patterns**: Recurring insights or learnings with effectiveness tracking
- **Knowledge**: Categorized information storage for facts and discoveries
### 3. Handoff Context
The handoff system provides incoming AIs with:
- Current session status and duration
- Active/pending tasks requiring attention
- Recent decisions that affect current work
- High-priority patterns that influence approach
- Project intelligence metrics
## Using the Handoff System
### For Incoming AI Assistants
**Step 1: Get Handoff Summary**
```bash
cmem handoff
```
This provides a markdown summary optimized for AI consumption with:
- Current session context
- Active tasks requiring attention
- Recent decisions influencing work
- Key patterns to apply
- Project statistics
**Step 2: Get Detailed Context**
```bash
cmem context
```
This provides JSON data with complete structured information:
- Full task details with IDs and metadata
- Complete decision history with alternatives
- Pattern effectiveness and frequency data
- Session tracking information
**Step 3: Continue or Start New Session**
Based on handoff information:
- If session < 4 hours old: Continue current session
- If session > 4 hours old: Auto-rotation will start new session
- Use `cmem session start "New Session Name"` for manual session creation
### For Outgoing AI Assistants
**Before Ending Work:**
1. **Complete Tasks**: Update any finished work
```bash
cmem task complete <task-id>
```
2. **Record Decisions**: Document important choices made
```bash
cmem decision "Decision" "Reasoning" "Alternative1,Alternative2"
```
3. **Add Patterns**: Capture learnings for future AIs
```bash
cmem pattern add "Pattern Name" "Description" --priority high
```
4. **Update Knowledge**: Store important discoveries
```bash
cmem knowledge add "key" "value" --category "category"
```
5. **End Session** (optional):
```bash
cmem session end "Outcome description"
```
## Integration with MCP Browser
The MCP Browser memory server automatically syncs with cmem when available:
### Automatic Sync
- **Task Operations**: Adding, updating, completing tasks sync to cmem
- **Pattern Creation**: New patterns are automatically added to cmem
- **Decision Recording**: Decisions made through MCP are stored in cmem
- **Identity-Based Storage**: Each identity gets separate memory space
### Identity System
```bash
# Use onboarding tool with identity-specific instructions
onboarding identity="ProjectName" instructions="Focus on code quality"
# Memory server uses identity for separate storage
# cmem integration syncs under that identity context
```
### Bidirectional Flow
1. **MCP → cmem**: Tool operations automatically sync to persistent storage
2. **cmem → MCP**: Memory server can read cmem data for context
3. **Cross-Session**: Patterns and decisions persist across AI instances
## Handoff Data Structure
### Session Information
```json
{
"session": {
"id": "2025-06-28-morning-development",
"name": "Morning Development",
"startTime": "2025-06-28T09:31:17.858Z",
"status": "active"
}
}
```
### Task Structure
```json
{
"id": "43801be2",
"description": "Task description",
"priority": "high|medium|low",
"status": "open|in_progress|completed",
"assignee": "assignee_name",
"createdAt": "2025-06-26T02:59:51.654Z"
}
```
### Decision Structure
```json
{
"id": "793cbd6e",
"decision": "Decision made",
"reasoning": "Why this was chosen",
"alternatives": ["Alt 1", "Alt 2"],
"timestamp": "2025-06-26T14:48:36.187Z"
}
```
### Pattern Structure
```json
{
"id": "03c8e07c",
"pattern": "Pattern Name",
"description": "Detailed description",
"priority": "high|medium|low",
"effectiveness": 0.8,
"frequency": 5
}
```
## Best Practices for AI Handoffs
### 1. **Read Before Acting**
Always check handoff information before starting work:
```bash
# Quick check
cmem handoff
# Detailed context for complex work
cmem context
```
### 2. **Maintain Context Continuity**
- Continue existing sessions when < 4 hours old
- Reference previous decisions in new work
- Apply high-priority patterns to current tasks
- Use established assignee names for consistency
### 3. **Document Decisions**
Record ANY significant choice:
- Technology selections
- Architecture decisions
- Approach changes
- Problem-solving strategies
### 4. **Pattern Recognition**
Capture insights that will help future AIs:
- Recurring problems and solutions
- Effective approaches
- Things to avoid
- Meta-patterns about the development process
### 5. **Task Management**
- Break large work into trackable tasks
- Update status as work progresses
- Complete tasks when finished
- Create new tasks for discovered work
## Example Handoff Workflow
### Incoming AI Workflow
```bash
# 1. Get handoff summary
cmem handoff
# 2. Check specific task details
cmem task list
# 3. Review recent patterns
cmem pattern list --priority high
# 4. Start work based on active tasks
# ... do work ...
# 5. Update progress
cmem task update <task-id> in_progress
```
### Outgoing AI Workflow
```bash
# 1. Complete finished tasks
cmem task complete <task-id>
# 2. Document decisions made
cmem decision "Use Docker for Firecrawl" "Simpler deployment" "Native install,VM"
# 3. Add learning patterns
cmem pattern add "Test all new features" "Always add tests before committing" --priority high
# 4. Create tasks for remaining work
cmem task add "Fix failing tests" --priority high --assignee next-ai
# 5. End session with outcome
cmem session end "Completed MCP browser enhancements with tests"
```
## Integration with Development Workflow
### Pre-Commit Checklist
- [ ] All tasks updated with current status
- [ ] New decisions documented with reasoning
- [ ] Patterns captured from development process
- [ ] Knowledge updated with discoveries
- [ ] Next tasks created for continuation
### Session Management
- **Short sessions (< 1 hour)**: Continue existing session
- **Medium sessions (1-4 hours)**: Continue or start new based on context
- **Long sessions (> 4 hours)**: Auto-rotation creates new session
### Cross-Project Context
- Use identity parameter for project-specific contexts
- Different projects maintain separate memory spaces
- Patterns can be shared across projects when relevant
## Error Handling
### When cmem is Unavailable
- MCP memory server gracefully degrades to local storage
- Sync attempts fail silently without breaking functionality
- Manual sync possible when cmem becomes available
### Memory Conflicts
- Sessions auto-rotate to prevent conflicts
- Task IDs are unique across sessions
- Patterns merge based on similarity detection
This handoff system ensures smooth AI transitions and maintains project continuity across multiple development sessions.

111
MCP_QUICK_REFERENCE.md Normal file
View File

@ -0,0 +1,111 @@
# MCP Quick Reference for Fresh Claude Instances
## 🔗 MCP Interface Access
**Status**: ✅ mcp-browser is configured and running
- **Claude Desktop****mcp-browser** → **27 tools from 7 servers**
- **Config**: `/home/claude/.claude/mcp-browser/config.yaml`
- **Tools**: Built-in servers (screen, memory, patterns, onboarding) + external (claude-code, brave-search, filesystem, github)
## 🚀 Quick MCP Commands
### Discovery (Context-Safe)
```python
# List all available tool names (27 total)
mcp_discover(jsonpath="$.tools[*].name")
# Find memory-related tools (regex support)
mcp_discover(jsonpath="$.tools[?(@.name =~ /memory|task|pattern/i)]")
# Get server information
mcp_discover(jsonpath="$.servers[*].name")
# Get claude-code tools specifically
mcp_discover(jsonpath="$.servers['claude-code'].tools[*].name")
```
### Tool Execution
```python
# Call any discovered tool
mcp_call(
method="tools/call",
params={
"name": "task_list",
"arguments": {"status": "pending"}
}
)
# Call claude-code tools
mcp_call(
method="tools/call",
params={
"name": "claude-code::read_file",
"arguments": {"path": "/path/to/file.py"}
}
)
```
## 🧠 Memory & Handoff System
### Get Current Context
```bash
cmem handoff # Quick context summary
cmem task list # Active tasks
cmem pattern list # Available patterns
```
### Essential Memory Tools (via MCP)
- `task_add` - Add new tasks with priority/assignee
- `task_update` - Update task status
- `memory_summary` - Get project overview
- `knowledge_add` - Store information
- `pattern_add` - Record learning patterns
## 🛠️ Built-in MCP Tools Available
**Screen Management (8 tools)**:
- `create_session`, `execute`, `peek`, `list_sessions`, `kill_session`
**Memory & Tasks (10 tools)**:
- `task_add`, `task_list`, `task_update`, `decision_add`, `pattern_add`
**Auto-Response Patterns (5 tools)**:
- `add_pattern`, `list_patterns`, `test_pattern`
**Identity & Onboarding (4 tools)**:
- `onboarding`, `onboarding_list`, `onboarding_export`
## 🎯 Current Project Context
**Location**: `/mnt/data/claude/claude` (bind mounted from `/home/claude`)
**Active Projects**:
- `mcp-browser` (✅ Working, integrated)
- `xilope` (🔄 In development - XDG config system needed)
**Memory Storage**: `/mnt/data/claude/claude/.mcp-memory/`
**cmem Wrapper**: `/mnt/data/claude/claude/bin/cmem``/usr/local/bin/cmem`
## 🔧 For Fresh Claude Instances
1. **Read this file** for MCP context
2. **Run `cmem handoff`** for session continuity
3. **Use `mcp_discover`** to explore available tools
4. **Check `CLAUDE.md`** for project-specific instructions
5. **Access Xilope onboarding**: `onboarding(identity="xilope_production")`
## 📚 Documentation Locations
- **Handoff Guide**: `/home/claude/claude-utils/mcp-browser/HANDOFF_INSTRUCTIONS.md`
- **Xilope Production**: `/home/claude/claude-utils/mcp-browser/mcp_servers/onboarding/xilope_production.md`
- **MCP Config**: `/home/claude/.claude/mcp-browser/config.yaml`
- **Memory Files**: `/mnt/data/claude/claude/.mcp-memory/default/`
## ⚡ Generate Complete API Docs
```bash
cd /home/claude/claude-utils/mcp-browser
python setup.py gen_apidoc
# Creates: mcp_api_documentation.json (27 tools, 7 servers documented)
```
This provides **complete MCP ecosystem access** with persistent memory across Claude sessions.

View File

@ -1,13 +1,17 @@
# Makefile for MCP Browser
# Created by AI for AI development
.PHONY: test install clean lint docs help
.PHONY: test install clean lint docs help run restart stop status
help:
@echo "MCP Browser Development Commands"
@echo "================================"
@echo "make test - Run all tests (unit + integration)"
@echo "make install - Install package in development mode"
@echo "make run - Install and restart daemon from home directory"
@echo "make restart - Stop and restart daemon (without install)"
@echo "make stop - Stop running daemon"
@echo "make status - Show daemon status"
@echo "make lint - Run code quality checks"
@echo "make docs - Generate AI-friendly documentation"
@echo "make clean - Remove build artifacts"
@ -45,4 +49,55 @@ format:
black mcp_browser/ mcp_servers/ tests/
typecheck:
mypy mcp_browser/ --ignore-missing-imports
mypy mcp_browser/ --ignore-missing-imports
# Daemon management targets
status:
@echo "=== MCP Browser Daemon Status ==="
@if [ -f /tmp/mcp-browser-1004/mcp-browser.pid ]; then \
pid=$$(cat /tmp/mcp-browser-1004/mcp-browser.pid); \
echo "PID file exists: $$pid"; \
if ps -p $$pid > /dev/null 2>&1; then \
echo "✓ Daemon is running (PID: $$pid)"; \
ps -fp $$pid; \
else \
echo "✗ Daemon not running (stale PID file)"; \
fi; \
else \
echo "✗ No PID file found"; \
fi
stop:
@echo "=== Stopping MCP Browser Daemon ==="
@if [ -f /tmp/mcp-browser-1004/mcp-browser.pid ]; then \
pid=$$(cat /tmp/mcp-browser-1004/mcp-browser.pid); \
echo "Stopping daemon (PID: $$pid)..."; \
if ps -p $$pid > /dev/null 2>&1; then \
kill -TERM $$pid; \
sleep 2; \
if ps -p $$pid > /dev/null 2>&1; then \
echo "Daemon still running, force killing..."; \
kill -KILL $$pid; \
fi; \
echo "✓ Daemon stopped"; \
else \
echo "Daemon was not running"; \
fi; \
rm -f /tmp/mcp-browser-1004/mcp-browser.pid; \
else \
echo "No PID file found"; \
fi
restart:
@echo "=== Restarting MCP Browser Daemon ==="
$(MAKE) stop
@echo "Starting daemon from home directory..."
@cd /mnt/data/claude/claude && nohup /mnt/data/claude/claude/.venv/bin/mcp-browser-daemon > /dev/null 2>&1 &
@sleep 3
@echo "Verifying daemon restart..."
$(MAKE) status
run:
@echo "=== Installing and Running MCP Browser Daemon ==="
$(MAKE) install
$(MAKE) restart

View File

@ -33,7 +33,8 @@ MCP Browser acts as a smart proxy between AI systems and MCP servers, providing:
- Configuration-driven server management
4. **Built-in Servers**
- **Screen**: GNU screen session management for persistent processes
- **Tmux**: tmux session management for persistent processes (default)
- **Screen**: GNU screen session management (legacy, disabled by default)
- **Memory**: Project memory, tasks, decisions, and knowledge management
- **Patterns**: Auto-response pattern management for automation
- **Onboarding**: Identity-aware onboarding for AI contexts
@ -53,7 +54,7 @@ mcp-browser/
│ └── config.py # Configuration management
├── mcp_servers/ # Built-in MCP servers
│ ├── base.py # Base server implementation
│ ├── screen/ # Screen session management
│ ├── screen/ # Session management (tmux + screen)
│ ├── memory/ # Memory and context management
│ ├── pattern_manager/ # Pattern automation
│ └── onboarding/ # Identity-aware onboarding
@ -85,7 +86,10 @@ pip install git+https://github.com/Xilope0/mcp-browser.git
## Quick Start
```bash
# Run interactive mode
# Run enhanced interactive mode (NEW!)
./mcp-browser-interactive
# Run basic interactive mode
mcp-browser
# Run as MCP server (for chaining)
@ -139,6 +143,39 @@ In sparse mode (default), only 3 tools are initially visible:
All other tools (potentially hundreds) are hidden but fully accessible through these meta-tools.
## Enhanced Interactive Mode
The new `./mcp-browser-interactive` provides a much better testing and exploration experience:
**Features:**
- **Tab completion** for commands and tool names
- **Command history** with readline support
- **Smart argument parsing** with key=value syntax
- **Built-in help** and tool discovery
- **Test mode** to try tools with sample data
- **Direct tool calls** without verbose JSON-RPC syntax
**Interactive Commands:**
```bash
help # Show available commands
list [pattern] # List tools (with optional filter)
discover <jsonpath> # Explore using JSONPath
call <tool> key=value # Call tool with arguments
test <tool> # Test tool with sample data
<tool> key=value # Direct tool call (shortcut)
onboard <identity> # Manage onboarding instructions
status # Show connection status
```
**Example Session:**
```bash
mcp> list bash # Find bash-related tools
mcp> discover $.tools[*].name # List all tool names
mcp> test Bash # Test Bash tool
mcp> Bash command="ls -la" # Direct tool call
mcp> onboard Claude "Focus on code quality" # Set onboarding
```
## Design Principles
1. **Generic**: No tool-specific knowledge built into the browser

View File

@ -3,6 +3,11 @@
```
./
setup.py
test_claude_desktop_flow.py
test_discovery.py
test_mcp_protocol.py
test_screen_multiuser.py
test_screen_utf8.py
mcp_browser.egg-info/
systemd/
build/
@ -51,7 +56,9 @@
test_brave_search.py
test_browser_functionality.py
test_claude_connection.py
test_cmem_integration.py
test_integration.py
test_interactive_client.py
test_onboarding.py
test_simple.py
config/
@ -80,6 +87,7 @@
daemon_main.py
default_configs.py
filter.py
interactive_client.py
logging_config.py
multi_server.py
proxy.py

16
mcp-browser-interactive Executable file
View File

@ -0,0 +1,16 @@
#!/usr/bin/env python3
"""
MCP Browser Interactive Mode Launcher
"""
import sys
import asyncio
from pathlib import Path
# Add the package to path
sys.path.insert(0, str(Path(__file__).parent))
from mcp_browser.interactive_client import main
if __name__ == "__main__":
asyncio.run(main())

165
mcp_api_documentation.json Normal file
View File

@ -0,0 +1,165 @@
{
"mcp_browser_version": "0.2.0",
"total_servers": 7,
"total_tools": 27,
"generation_timestamp": "2025-06-28T13:16:38.436458",
"servers": {
"default": {
"name": "default",
"description": "Default in-memory MCP server",
"command": [
"npx",
"-y",
"@modelcontextprotocol/server-memory"
],
"status": "configured",
"tools": [],
"tool_count": 0,
"tool_names": [],
"environment": {},
"working_directory": null,
"capabilities": []
},
"claude-code": {
"name": "claude-code",
"description": "MCP interface to claude code started in /home/claude",
"command": [
"/home/claude/bin/claude"
],
"status": "configured",
"tools": [],
"tool_count": 0,
"tool_names": [],
"environment": {},
"working_directory": null,
"capabilities": []
},
"brave-search": {
"name": "brave-search",
"description": "Brave Search API access",
"command": [
"npx",
"-y",
"@modelcontextprotocol/server-brave-search"
],
"status": "configured",
"tools": [],
"tool_count": 0,
"tool_names": [],
"environment": {},
"working_directory": null,
"capabilities": []
},
"filesystem": {
"name": "filesystem",
"description": "File system access (/home/claude)",
"command": [
"npx",
"-y",
"@modelcontextprotocol/server-filesystem",
"/home/claude"
],
"status": "configured",
"tools": [],
"tool_count": 0,
"tool_names": [],
"environment": {},
"working_directory": null,
"capabilities": []
},
"github": {
"name": "github",
"description": "GitHub API access",
"command": [
"npx",
"-y",
"@modelcontextprotocol/server-github"
],
"status": "configured",
"tools": [],
"tool_count": 0,
"tool_names": [],
"environment": {},
"working_directory": null,
"capabilities": []
},
"memory": {
"name": "memory",
"description": "Persistent memory and notes",
"command": [
"npx",
"-y",
"@modelcontextprotocol/server-memory"
],
"status": "configured",
"tools": [],
"tool_count": 0,
"tool_names": [],
"environment": {},
"working_directory": null,
"capabilities": []
},
"builtin-only": {
"name": "builtin-only",
"description": "Use only built-in Python servers",
"command": null,
"status": "configured",
"tools": [],
"tool_count": 0,
"tool_names": [],
"environment": {},
"working_directory": null,
"capabilities": []
}
},
"builtin": {
"name": "builtin",
"description": "Built-in MCP Browser servers (screen, memory, patterns, onboarding)",
"status": "active",
"tools": [],
"tool_count": 0,
"capabilities": [
"screen_management",
"memory_storage",
"pattern_matching",
"onboarding_management"
]
},
"discovery_patterns": {
"all_tools": "$.tools[*]",
"all_tool_names": "$.tools[*].name",
"tools_by_server": "$.servers[*].tools[*]",
"tool_schemas": "$.tools[*].inputSchema",
"memory_tools": "$.tools[?(@.name =~ /memory|task|pattern|knowledge/i)]",
"screen_tools": "$.tools[?(@.name =~ /screen|session/i)]",
"find_tool_by_name": "$.tools[?(@.name=='TOOL_NAME')]",
"server_capabilities": "$.servers[*].capabilities"
},
"sparse_mode_info": {
"visible_tools": [
"mcp_discover",
"mcp_call",
"onboarding"
],
"hidden_tools": 27,
"purpose": "Context optimization - full MCP API accessible via proxy tools"
},
"runtime_status": {
"builtin:screen": {
"status": "active",
"pid": 589886
},
"builtin:memory": {
"status": "active",
"pid": 589890
},
"builtin:patterns": {
"status": "active",
"pid": 589905
},
"builtin:onboarding": {
"status": "active",
"pid": 589908
}
}
}

View File

@ -14,7 +14,7 @@ the Free Software Foundation, either version 3 of the License, or
from .proxy import MCPBrowser
__version__ = "0.1.0"
__version__ = "0.2.0"
__author__ = "Claude4Ξlope"
__email__ = "xilope@esus.name"
__license__ = "GPLv3+"

View File

@ -22,6 +22,7 @@ class MCPServerConfig:
env: Dict[str, str] = field(default_factory=dict)
name: Optional[str] = None
description: Optional[str] = None
enabled: bool = True
@dataclass
@ -88,7 +89,8 @@ class ConfigLoader:
args=server_config.get("args", []),
env=server_config.get("env", {}),
name=server_config.get("name", name),
description=server_config.get("description")
description=server_config.get("description"),
enabled=server_config.get("enabled", True)
)
self._config = MCPBrowserConfig(

View File

@ -0,0 +1,523 @@
#!/usr/bin/env python3
"""
Enhanced Interactive MCP Browser Client
Provides a user-friendly interactive interface for exploring and using MCP tools
with better discovery, autocompletion, and testing capabilities.
"""
import asyncio
import json
import readline
import sys
from pathlib import Path
from typing import Dict, Any, List, Optional
import traceback
from .proxy import MCPBrowser
from .daemon import MCPBrowserClient, get_socket_path, is_daemon_running
from .logging_config import get_logger
class InteractiveMCPClient:
"""Enhanced interactive MCP browser client."""
def __init__(self, server_name: Optional[str] = None, use_daemon: bool = True):
self.server_name = server_name
self.use_daemon = use_daemon
self.browser: Optional[MCPBrowser] = None
self.client: Optional[MCPBrowserClient] = None
self.logger = get_logger(__name__)
self.tool_cache: Dict[str, Any] = {}
self.command_history: List[str] = []
# Setup readline
self._setup_readline()
def _setup_readline(self):
"""Setup readline for better command line experience."""
readline.set_completer(self._completer)
readline.parse_and_bind('tab: complete')
readline.set_completer_delims(' \t\n`!@#$%^&*()=+[{]}\\|;:\'",<>?')
# Load history
history_file = Path.home() / ".mcp_browser_history"
try:
readline.read_history_file(str(history_file))
except FileNotFoundError:
pass
# Save history on exit
import atexit
atexit.register(readline.write_history_file, str(history_file))
def _completer(self, text: str, state: int) -> Optional[str]:
"""Tab completion for commands and tool names."""
if state == 0:
# Get current line
line = readline.get_line_buffer()
# Complete commands
commands = ['discover', 'call', 'list', 'help', 'quit', 'onboard', 'status', 'test']
# Add tool names if we have them cached
if self.tool_cache:
commands.extend(self.tool_cache.keys())
# Filter matches
self.matches = [cmd for cmd in commands if cmd.startswith(text)]
try:
return self.matches[state]
except IndexError:
return None
async def initialize(self):
"""Initialize the MCP browser connection."""
print("🔍 MCP Browser Interactive Mode")
print("Type 'help' for commands, 'quit' to exit")
print()
# Try to connect
if self.use_daemon:
socket_path = get_socket_path(self.server_name)
if is_daemon_running(socket_path):
try:
self.client = MCPBrowserClient(socket_path)
await self.client.__aenter__()
print(f"✅ Connected to daemon at {socket_path}")
except Exception as e:
self.logger.warning(f"Failed to connect to daemon: {e}")
self.client = None
if not self.client:
# Fallback to standalone
print("🚀 Starting standalone MCP browser...")
self.browser = MCPBrowser(server_name=self.server_name)
await self.browser.initialize()
print("✅ MCP browser initialized")
# Load initial tool list
await self._refresh_tools()
async def _refresh_tools(self):
"""Refresh the tool cache."""
try:
response = await self._call_mcp({
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list",
"params": {}
})
if "result" in response and "tools" in response["result"]:
self.tool_cache.clear()
for tool in response["result"]["tools"]:
self.tool_cache[tool["name"]] = tool
except Exception as e:
self.logger.warning(f"Failed to refresh tools: {e}")
async def _call_mcp(self, request: Dict[str, Any]) -> Dict[str, Any]:
"""Call MCP method through client or browser."""
if self.client:
return await self.client.call(request)
elif self.browser:
return await self.browser.call(request)
else:
raise RuntimeError("No MCP connection available")
async def run(self):
"""Main interactive loop."""
try:
await self.initialize()
while True:
try:
# Get user input
line = input("mcp> ").strip()
if not line:
continue
self.command_history.append(line)
# Parse and execute command
await self._execute_command(line)
except KeyboardInterrupt:
print("\nUse 'quit' to exit")
continue
except EOFError:
break
except Exception as e:
print(f"❌ Error: {e}")
traceback.print_exc()
finally:
await self.cleanup()
async def _execute_command(self, line: str):
"""Execute a user command."""
parts = line.split()
if not parts:
return
command = parts[0].lower()
args = parts[1:]
if command == 'help':
self._show_help()
elif command == 'quit' or command == 'exit':
print("👋 Goodbye!")
sys.exit(0)
elif command == 'list':
await self._list_tools(args)
elif command == 'discover':
await self._discover_tools(args)
elif command == 'call':
await self._call_tool(args)
elif command == 'onboard':
await self._manage_onboarding(args)
elif command == 'status':
await self._show_status()
elif command == 'test':
await self._test_tool(args)
elif command == 'refresh':
await self._refresh_tools()
print("🔄 Tool cache refreshed")
else:
# Try to call it as a tool directly
await self._call_tool_direct(command, args)
def _show_help(self):
"""Show help information."""
help_text = """
🔍 MCP Browser Interactive Commands
Basic Commands:
help Show this help
quit, exit Exit the browser
refresh Refresh tool cache
status Show connection status
Tool Discovery:
list [pattern] List available tools (optional filter)
discover <jsonpath> Discover tools using JSONPath
Tool Execution:
call <tool> [args...] Call a tool with arguments
test <tool> Test a tool with sample data
<tool> [args...] Direct tool call (shortcut)
Onboarding:
onboard <identity> Get onboarding for identity
onboard <identity> <instructions> Set onboarding
Examples:
list # List all tools
list bash # List tools containing 'bash'
discover $.tools[*].name # Get all tool names
discover $.tools[?(@.name=='Bash')] # Get Bash tool details
call mcp_discover jsonpath="$.tools[*].name"
test Bash # Test Bash tool
onboard Claude # Get Claude's onboarding
onboard Claude "Focus on code quality" # Set onboarding
"""
print(help_text)
async def _list_tools(self, args: List[str]):
"""List available tools with optional filtering."""
pattern = args[0] if args else None
tools = list(self.tool_cache.values())
if pattern:
tools = [t for t in tools if pattern.lower() in t["name"].lower() or
pattern.lower() in t.get("description", "").lower()]
if not tools:
print("❌ No tools found")
return
print(f"📋 Available Tools ({len(tools)} found):")
print()
for tool in tools:
name = tool["name"]
desc = tool.get("description", "No description")
# Truncate long descriptions
if len(desc) > 80:
desc = desc[:77] + "..."
# Add emoji based on tool type
emoji = "🔍" if "discover" in name else "🚀" if "call" in name else "📋" if "onboard" in name else "🛠️"
print(f" {emoji} {name}")
print(f" {desc}")
print()
async def _discover_tools(self, args: List[str]):
"""Discover tools using JSONPath."""
if not args:
print("❌ Usage: discover <jsonpath>")
print("Examples:")
print(" discover $.tools[*].name")
print(" discover $.tools[?(@.name=='Bash')]")
return
jsonpath = " ".join(args)
try:
response = await self._call_mcp({
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "mcp_discover",
"arguments": {"jsonpath": jsonpath}
}
})
if "result" in response:
content = response["result"].get("content", [])
if content and content[0].get("type") == "text":
result = content[0]["text"]
print("🔍 Discovery Result:")
print(result)
else:
print("❌ No content in response")
elif "error" in response:
print(f"❌ Error: {response['error']['message']}")
except Exception as e:
print(f"❌ Discovery failed: {e}")
async def _call_tool(self, args: List[str]):
"""Call a tool with arguments."""
if not args:
print("❌ Usage: call <tool_name> [key=value...]")
print("Example: call mcp_discover jsonpath=\"$.tools[*].name\"")
return
tool_name = args[0]
# Parse key=value arguments
arguments = {}
for arg in args[1:]:
if "=" in arg:
key, value = arg.split("=", 1)
# Remove quotes if present
value = value.strip('"\'')
arguments[key] = value
else:
# Positional argument - try to guess the parameter name
if tool_name in self.tool_cache:
tool = self.tool_cache[tool_name]
schema = tool.get("inputSchema", {})
props = schema.get("properties", {})
required = schema.get("required", [])
# Use first required parameter
if required and len(arguments) == 0:
arguments[required[0]] = arg
else:
arguments[f"arg_{len(arguments)}"] = arg
await self._execute_tool_call(tool_name, arguments)
async def _call_tool_direct(self, tool_name: str, args: List[str]):
"""Direct tool call (shortcut syntax)."""
if tool_name not in self.tool_cache:
print(f"❌ Unknown tool: {tool_name}")
print("Use 'list' to see available tools")
return
# Parse arguments like _call_tool
arguments = {}
for arg in args:
if "=" in arg:
key, value = arg.split("=", 1)
value = value.strip('"\'')
arguments[key] = value
else:
# Use tool schema to guess parameter
tool = self.tool_cache[tool_name]
schema = tool.get("inputSchema", {})
required = schema.get("required", [])
if required and len(arguments) == 0:
arguments[required[0]] = arg
await self._execute_tool_call(tool_name, arguments)
async def _execute_tool_call(self, tool_name: str, arguments: Dict[str, Any]):
"""Execute a tool call and display results."""
print(f"🚀 Calling {tool_name} with {arguments}")
try:
response = await self._call_mcp({
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": arguments
}
})
if "result" in response:
self._display_result(response["result"])
elif "error" in response:
print(f"❌ Error: {response['error']['message']}")
except Exception as e:
print(f"❌ Tool call failed: {e}")
def _display_result(self, result: Any):
"""Display tool call result in a nice format."""
if isinstance(result, dict) and "content" in result:
# MCP content format
content = result["content"]
for item in content:
if item.get("type") == "text":
print("📄 Result:")
print(item["text"])
elif item.get("type") == "image":
print(f"🖼️ Image: {item.get('url', 'No URL')}")
else:
print(f"📦 Content: {json.dumps(item, indent=2)}")
else:
# Raw result
print("📦 Result:")
if isinstance(result, (dict, list)):
print(json.dumps(result, indent=2))
else:
print(str(result))
async def _test_tool(self, args: List[str]):
"""Test a tool with sample data."""
if not args:
print("❌ Usage: test <tool_name>")
return
tool_name = args[0]
if tool_name not in self.tool_cache:
print(f"❌ Unknown tool: {tool_name}")
return
tool = self.tool_cache[tool_name]
schema = tool.get("inputSchema", {})
print(f"🧪 Testing {tool_name}")
print(f"📋 Description: {tool.get('description', 'No description')}")
print(f"📊 Schema: {json.dumps(schema, indent=2)}")
# Generate sample arguments
sample_args = self._generate_sample_args(schema)
print(f"🎲 Sample arguments: {sample_args}")
# Ask user if they want to proceed
try:
confirm = input("Proceed with test? [y/N]: ").strip().lower()
if confirm in ['y', 'yes']:
await self._execute_tool_call(tool_name, sample_args)
except KeyboardInterrupt:
print("\n❌ Test cancelled")
def _generate_sample_args(self, schema: Dict[str, Any]) -> Dict[str, Any]:
"""Generate sample arguments based on schema."""
args = {}
props = schema.get("properties", {})
for name, prop in props.items():
prop_type = prop.get("type", "string")
if prop_type == "string":
if "example" in prop:
args[name] = prop["example"]
elif name.lower() in ["jsonpath", "path"]:
args[name] = "$.tools[*].name"
elif name.lower() in ["query", "search"]:
args[name] = "test query"
else:
args[name] = f"sample_{name}"
elif prop_type == "boolean":
args[name] = False
elif prop_type == "number":
args[name] = 1
elif prop_type == "array":
args[name] = ["sample"]
elif prop_type == "object":
args[name] = {}
return args
async def _manage_onboarding(self, args: List[str]):
"""Manage onboarding instructions."""
if not args:
print("❌ Usage: onboard <identity> [instructions]")
return
identity = args[0]
instructions = " ".join(args[1:]) if len(args) > 1 else None
arguments = {"identity": identity}
if instructions:
arguments["instructions"] = instructions
await self._execute_tool_call("onboarding", arguments)
async def _show_status(self):
"""Show connection and tool status."""
print("📊 MCP Browser Status")
print()
if self.client:
print("🔗 Connection: Daemon")
elif self.browser:
print("🔗 Connection: Standalone")
else:
print("❌ Connection: None")
print(f"🛠️ Tools cached: {len(self.tool_cache)}")
print(f"📝 Command history: {len(self.command_history)}")
if self.server_name:
print(f"🎯 Server: {self.server_name}")
# Show tool breakdown
if self.tool_cache:
meta_tools = [name for name in self.tool_cache if name.startswith("mcp_") or name == "onboarding"]
regular_tools = [name for name in self.tool_cache if name not in meta_tools]
print()
print(f"🔍 Meta tools: {len(meta_tools)} ({', '.join(meta_tools)})")
print(f"🛠️ Regular tools: {len(regular_tools)}")
async def cleanup(self):
"""Cleanup resources."""
try:
if self.client:
await self.client.__aexit__(None, None, None)
if self.browser:
await self.browser.close()
except Exception as e:
self.logger.warning(f"Cleanup error: {e}")
async def main():
"""Main entry point for interactive mode."""
import argparse
parser = argparse.ArgumentParser(description="Interactive MCP Browser")
parser.add_argument("--server", help="MCP server name")
parser.add_argument("--no-daemon", action="store_true", help="Don't use daemon")
args = parser.parse_args()
client = InteractiveMCPClient(
server_name=args.server,
use_daemon=not args.no_daemon
)
await client.run()
if __name__ == "__main__":
asyncio.run(main())

View File

@ -28,10 +28,16 @@ class MultiServerManager:
base_path = Path(__file__).parent.parent / "mcp_servers"
return {
"builtin:tmux": MCPServerConfig(
command=["python3", str(base_path / "screen" / "tmux_server.py")],
name="tmux",
description="tmux session management"
),
"builtin:screen": MCPServerConfig(
command=["python3", str(base_path / "screen" / "screen_server.py")],
name="screen",
description="GNU screen session management"
description="GNU screen session management (legacy)",
enabled=False # Disabled by default, tmux is preferred
),
"builtin:memory": MCPServerConfig(
command=["python3", str(base_path / "memory" / "memory_server.py")],
@ -53,6 +59,11 @@ class MultiServerManager:
async def start_builtin_servers(self):
"""Start all built-in servers."""
for name, config in self.builtin_servers.items():
# Skip disabled servers
if not config.enabled:
self.logger.info(f"Skipping disabled built-in server: {name}")
continue
self.logger.info(f"Starting built-in server: {name}")
server = MCPServer(config, logger=get_logger(__name__, name))

View File

@ -6,6 +6,7 @@ and supports sparse mode for context optimization.
"""
import json
import re
from typing import Dict, Any, List, Optional, Union
from jsonpath_ng import parse as jsonpath_parse
from jsonpath_ng.exceptions import JsonPathParserError
@ -56,6 +57,10 @@ class ToolRegistry:
$.tools[?(@.name=='Bash')] - Get Bash tool details
$.tools[*].inputSchema - Get all input schemas
"""
# Check if this is a regex query and handle it specially
if "=~" in jsonpath:
return self._regex_search(jsonpath)
try:
expr = jsonpath_parse(jsonpath)
except (JsonPathParserError, Exception):
@ -79,6 +84,89 @@ class ToolRegistry:
else:
return [match.value for match in matches]
def _regex_search(self, jsonpath: str) -> Union[List[Any], Any, None]:
"""
Handle regex-based JSONPath queries manually.
Supports patterns like: $.tools[?(@.name =~ /pattern/flags)]
"""
# Parse basic regex patterns for tools
if "$.tools[?(@.name =~" in jsonpath:
# Extract regex pattern
match = re.search(r'/([^/]+)/([gi]*)', jsonpath)
if not match:
return None
pattern = match.group(1)
flags_str = match.group(2)
# Convert flags
flags = 0
if 'i' in flags_str:
flags |= re.IGNORECASE
if 'g' in flags_str:
pass # Global is default behavior in Python findall
try:
regex = re.compile(pattern, flags)
except re.error:
return None
# Search through tools
matches = []
for tool in self.raw_tool_list:
tool_name = tool.get("name", "")
if regex.search(tool_name):
matches.append(tool)
return matches if matches else None
elif "$.tools[?(@.description =~" in jsonpath:
# Extract regex pattern for descriptions
match = re.search(r'/([^/]+)/([gi]*)', jsonpath)
if not match:
return None
pattern = match.group(1)
flags_str = match.group(2)
flags = 0
if 'i' in flags_str:
flags |= re.IGNORECASE
try:
regex = re.compile(pattern, flags)
except re.error:
return None
# Search through tool descriptions
matches = []
for tool in self.raw_tool_list:
description = tool.get("description", "")
if regex.search(description):
matches.append(tool)
return matches if matches else None
# Fallback to basic JSONPath if regex pattern not recognized
try:
expr = jsonpath_parse(jsonpath.replace("=~", "==")) # Try basic equality
search_data = {
"tools": self.raw_tool_list,
"tool_names": self.get_all_tool_names(),
"metadata": self._metadata,
"servers": self._metadata.get("servers", {})
}
matches = expr.find(search_data)
if not matches:
return None
elif len(matches) == 1:
return matches[0].value
else:
return [match.value for match in matches]
except:
return None
def get_sparse_tools(self) -> List[Dict[str, Any]]:
"""
Get minimal tool list for sparse mode.
@ -91,13 +179,13 @@ class ToolRegistry:
sparse_tools = [
{
"name": "mcp_discover",
"description": f"Discover available tools and servers using JSONPath. {tool_count} tools from {server_count} servers available.",
"description": f"🔍 PROXY META-TOOL: Discover {tool_count} hidden tools from {server_count} MCP servers without loading them into context. This prevents context explosion while enabling full tool access via JSONPath queries. Use this to explore what's available before calling specific tools.",
"inputSchema": {
"type": "object",
"properties": {
"jsonpath": {
"type": "string",
"description": "JSONPath expression (e.g., '$.tools[*].name')"
"description": "JSONPath expression to query tool catalog. Examples: '$.tools[*].name' (list all), '$.tools[?(@.name=='Bash')]' (find specific), '$.servers[*]' (list servers)"
}
},
"required": ["jsonpath"]
@ -105,17 +193,17 @@ class ToolRegistry:
},
{
"name": "mcp_call",
"description": "Execute any MCP tool by constructing a JSON-RPC call.",
"description": f"🚀 PROXY META-TOOL: Execute any of the {tool_count} available MCP tools by constructing JSON-RPC calls. This is the universal interface to all hidden tools - you can call ANY tool discovered via mcp_discover without it being loaded into your context.",
"inputSchema": {
"type": "object",
"properties": {
"method": {
"type": "string",
"description": "JSON-RPC method (e.g., 'tools/call')"
"description": "JSON-RPC method to call. For tool execution use 'tools/call'. Other methods: 'tools/list', 'prompts/list', 'resources/list'"
},
"params": {
"type": "object",
"description": "Method parameters"
"description": "Method parameters. For 'tools/call': {'name': 'tool_name', 'arguments': {...}}. The arguments object contains the actual tool parameters."
}
},
"required": ["method", "params"]
@ -123,21 +211,21 @@ class ToolRegistry:
},
{
"name": "onboarding",
"description": "Get or set identity-specific onboarding instructions for AI contexts.",
"description": "📋 BUILT-IN TOOL: Manage persistent, identity-aware onboarding instructions. This tool lets AI instances leave instructions for future contexts based on identity (project name, user, etc). Perfect for maintaining context across sessions without consuming tokens.",
"inputSchema": {
"type": "object",
"properties": {
"identity": {
"type": "string",
"description": "Identity for onboarding (e.g., 'Claude', project name)"
"description": "Identity key for onboarding instructions (e.g., 'Claude', 'MyProject', 'WebDev'). Each identity can have separate instructions."
},
"instructions": {
"type": "string",
"description": "Optional: Set new instructions. If omitted, retrieves existing."
"description": "Optional: New instructions to store. If omitted, retrieves existing instructions for this identity. Use this to leave notes for future AI sessions."
},
"append": {
"type": "boolean",
"description": "Append to existing instructions instead of replacing",
"description": "If true, append to existing instructions instead of replacing them entirely",
"default": False
}
},
@ -148,8 +236,126 @@ class ToolRegistry:
return sparse_tools
def set_metadata(self, key: str, value: Any):
"""Set metadata that can be discovered via JSONPath."""
def get_full_api_documentation(self) -> Dict[str, Any]:
"""
Generate comprehensive API documentation for AI consumption.
Returns complete server and tool information in structured JSON format.
"""
servers = self._metadata.get("servers", {})
# Group tools by server
tools_by_server = {}
builtin_tools = []
for tool in self.raw_tool_list:
tool_name = tool.get("name", "")
# Check if it's a server-namespaced tool
if "::" in tool_name:
server_ns = tool_name.split("::")[0]
if server_ns not in tools_by_server:
tools_by_server[server_ns] = []
tools_by_server[server_ns].append(tool)
else:
# Check if tool belongs to a specific server based on metadata
found_server = None
for server_name, server_info in servers.items():
server_tools = server_info.get("tools", [])
if any(t.get("name") == tool_name for t in server_tools):
found_server = server_name
break
if found_server:
if found_server not in tools_by_server:
tools_by_server[found_server] = []
tools_by_server[found_server].append(tool)
else:
builtin_tools.append(tool)
# Build comprehensive documentation
api_doc = {
"mcp_browser_version": "0.2.0",
"total_servers": len(servers) + (1 if builtin_tools else 0),
"total_tools": len(self.raw_tool_list),
"generation_timestamp": None, # Will be set when called
"servers": {},
"builtin": {
"name": "builtin",
"description": "Built-in MCP Browser servers (screen, memory, patterns, onboarding)",
"status": "active",
"tools": builtin_tools,
"tool_count": len(builtin_tools),
"capabilities": ["screen_management", "memory_storage", "pattern_matching", "onboarding_management"]
},
"discovery_patterns": {
"all_tools": "$.tools[*]",
"all_tool_names": "$.tools[*].name",
"tools_by_server": "$.servers[*].tools[*]",
"tool_schemas": "$.tools[*].inputSchema",
"memory_tools": "$.tools[?(@.name =~ /memory|task|pattern|knowledge/i)]",
"screen_tools": "$.tools[?(@.name =~ /screen|session/i)]",
"find_tool_by_name": "$.tools[?(@.name=='TOOL_NAME')]",
"server_capabilities": "$.servers[*].capabilities"
},
"sparse_mode_info": {
"visible_tools": ["mcp_discover", "mcp_call", "onboarding"],
"hidden_tools": len(self.raw_tool_list),
"purpose": "Context optimization - full MCP API accessible via proxy tools"
}
}
# Add external servers
for server_name, server_info in servers.items():
server_tools = tools_by_server.get(server_name, [])
api_doc["servers"][server_name] = {
"name": server_name,
"description": server_info.get("description", ""),
"command": server_info.get("command", []),
"status": server_info.get("status", "unknown"),
"tools": server_tools,
"tool_count": len(server_tools),
"tool_names": [t.get("name", "") for t in server_tools],
"environment": server_info.get("env", {}),
"working_directory": server_info.get("cwd"),
"capabilities": self._extract_capabilities(server_tools)
}
return api_doc
def _extract_capabilities(self, tools: List[Dict[str, Any]]) -> List[str]:
"""Extract capabilities from tool list."""
capabilities = set()
for tool in tools:
name = tool.get("name", "").lower()
desc = tool.get("description", "").lower()
# Infer capabilities from tool names and descriptions
if any(keyword in name for keyword in ["read", "write", "file"]):
capabilities.add("file_operations")
if any(keyword in name for keyword in ["search", "query", "find"]):
capabilities.add("search_operations")
if any(keyword in name for keyword in ["web", "http", "url"]):
capabilities.add("web_operations")
if any(keyword in name for keyword in ["git", "repo", "commit"]):
capabilities.add("version_control")
if any(keyword in name for keyword in ["memory", "store", "save"]):
capabilities.add("data_storage")
if any(keyword in name for keyword in ["exec", "run", "command"]):
capabilities.add("command_execution")
if any(keyword in desc for keyword in ["browser", "scrape", "crawl"]):
capabilities.add("web_scraping")
return sorted(list(capabilities))
def set_metadata(self, metadata: Dict[str, Any]):
"""Set metadata about servers and configuration."""
self._metadata = metadata
def update_metadata(self, key: str, value: Any):
"""Set specific metadata that can be discovered via JSONPath."""
self._metadata[key] = value
def to_json(self) -> str:

View File

@ -54,13 +54,14 @@ class Pattern:
class MemoryServer(BaseMCPServer):
"""MCP server for memory and context management."""
"""MCP server for memory and context management with cmem integration."""
def __init__(self):
def __init__(self, identity: str = "default"):
super().__init__("memory-server", "1.0.0")
self.memory_dir = Path.home() / ".mcp-memory"
self.memory_dir.mkdir(exist_ok=True)
self.current_project = "default"
self.current_project = identity
self.cmem_integration = self._setup_cmem_integration()
self._register_tools()
self._load_memory()
@ -211,6 +212,53 @@ class MemoryServer(BaseMCPServer):
self.patterns = self._load_json("patterns.json", {})
self.knowledge = self._load_json("knowledge.json", {})
def _setup_cmem_integration(self) -> bool:
"""Setup integration with cmem by creating identity-specific directories."""
try:
# Check if cmem is available
import subprocess
result = subprocess.run(['cmem', 'stats'], capture_output=True, text=True, timeout=5)
if result.returncode != 0:
return False
# Create identity-specific directory
identity_dir = self.memory_dir / self.current_project
identity_dir.mkdir(exist_ok=True)
# Check if we should symlink to cmem storage
claude_dir = Path.home() / ".claude"
if claude_dir.exists():
# Try to find cmem session data
cmem_session_dirs = list(claude_dir.glob("sessions/*/"))
if cmem_session_dirs:
# Use the most recent session
latest_session = max(cmem_session_dirs, key=lambda p: p.stat().st_mtime)
# Create symlinks for task/pattern/decision integration
self._create_cmem_bridges(identity_dir, latest_session)
return True
return False
except Exception as e:
# Fail silently - cmem integration is optional
return False
def _create_cmem_bridges(self, identity_dir: Path, session_dir: Path):
"""Create bridge files to sync with cmem."""
# Create bridge files that can sync with cmem format
bridge_dir = identity_dir / "cmem_bridge"
bridge_dir.mkdir(exist_ok=True)
# Store reference to cmem session for potential sync
bridge_info = {
"session_dir": str(session_dir),
"last_sync": datetime.now().isoformat(),
"integration_active": True
}
with open(bridge_dir / "info.json", 'w') as f:
json.dump(bridge_info, f, indent=2)
def _load_json(self, filename: str, default: Any) -> Any:
"""Load JSON file or return default."""
filepath = self.project_dir / filename
@ -263,6 +311,9 @@ class MemoryServer(BaseMCPServer):
self.tasks[task.id] = asdict(task)
self._save_json("tasks.json", self.tasks)
# Try to sync with cmem if integration is active
await self._sync_task_to_cmem(task, "add")
return self.content_text(f"Added task: {task.id[:8]} - {task.content}")
async def _task_list(self, args: Dict[str, Any]) -> Dict[str, Any]:
@ -298,6 +349,9 @@ class MemoryServer(BaseMCPServer):
self.tasks[full_id]["status"] = new_status
if new_status == "completed":
self.tasks[full_id]["completed_at"] = datetime.now().isoformat()
# Sync completion to cmem
task_obj = Task(**self.tasks[full_id])
await self._sync_task_to_cmem(task_obj, "complete")
self._save_json("tasks.json", self.tasks)
@ -315,6 +369,9 @@ class MemoryServer(BaseMCPServer):
self.decisions[decision.id] = asdict(decision)
self._save_json("decisions.json", self.decisions)
# Try to sync with cmem
await self._sync_decision_to_cmem(decision)
return self.content_text(f"Recorded decision: {decision.choice}")
async def _pattern_add(self, args: Dict[str, Any]) -> Dict[str, Any]:
@ -330,6 +387,9 @@ class MemoryServer(BaseMCPServer):
self.patterns[pattern.id] = asdict(pattern)
self._save_json("patterns.json", self.patterns)
# Try to sync with cmem
await self._sync_pattern_to_cmem(pattern, "add")
return self.content_text(f"Added pattern: {pattern.pattern}")
async def _pattern_resolve(self, args: Dict[str, Any]) -> Dict[str, Any]:
@ -438,6 +498,89 @@ Total Knowledge Items: {sum(len(items) for items in self.knowledge.values())}
"""
return self.content_text(summary)
async def _sync_task_to_cmem(self, task: Task, action: str):
"""Sync task with cmem if integration is active."""
if not self.cmem_integration:
return
try:
import subprocess
import asyncio
if action == "add":
# Map our priority to cmem priority
priority_map = {"low": "low", "medium": "medium", "high": "high"}
cmem_priority = priority_map.get(task.priority, "medium")
# Add task to cmem
cmd = ['cmem', 'task', 'add', task.content, '--priority', cmem_priority]
if task.assignee:
cmd.extend(['--assignee', task.assignee])
result = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
await result.communicate()
elif action == "complete":
# Try to find and complete corresponding cmem task
# This is best-effort since we don't have direct ID mapping
cmd = ['cmem', 'task', 'complete', task.content[:50]] # Use content prefix
result = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
await result.communicate()
except Exception:
# Fail silently - cmem sync is optional
pass
async def _sync_pattern_to_cmem(self, pattern: Pattern, action: str):
"""Sync pattern with cmem if integration is active."""
if not self.cmem_integration:
return
try:
import subprocess
import asyncio
if action == "add":
# Add pattern to cmem
cmd = ['cmem', 'pattern', 'add', pattern.pattern, pattern.description]
if pattern.priority != "medium":
cmd.extend(['--priority', pattern.priority])
result = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
await result.communicate()
except Exception:
# Fail silently - cmem sync is optional
pass
async def _sync_decision_to_cmem(self, decision: Decision):
"""Sync decision with cmem if integration is active."""
if not self.cmem_integration:
return
try:
import subprocess
import asyncio
# Add decision to cmem
alternatives_str = ', '.join(decision.alternatives)
cmd = ['cmem', 'decision', decision.choice, decision.reasoning, alternatives_str]
result = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
await result.communicate()
except Exception:
# Fail silently - cmem sync is optional
pass
if __name__ == "__main__":

View File

@ -1,30 +1,62 @@
# MCP Browser - Universal Model Context Protocol Proxy
Welcome to MCP Browser! This tool acts as a proxy between AI systems and MCP servers, providing:
Welcome to MCP Browser! This tool solves the **context explosion problem** by acting as a smart proxy between AI systems and potentially hundreds of MCP tools.
## Core Capabilities
## The Context Problem MCP Browser Solves
### 1. **Proxy Mode**
MCP Browser acts as a transparent proxy to external MCP servers configured in `~/.claude/mcp-browser/config.yaml`. You can:
- Connect to any MCP server (filesystem, brave-search, github, etc.)
- Add new servers at runtime without restarting
- Access all tools from configured servers through the proxy
Traditional MCP setups expose ALL tools to the AI context immediately, which can easily consume thousands of tokens. MCP Browser implements a **minimal-to-maximal interface pattern**:
### 2. **Built-in Tools**
Always available, regardless of external servers:
- **Screen Management** - Create/manage GNU screen sessions
- **Memory & Tasks** - Persistent memory and task tracking
- **Pattern Manager** - Auto-response patterns
- **Onboarding** - Context-specific instructions (this tool)
- **What AI sees**: Only 3 simple meta-tools (minimal context usage)
- **What AI can access**: All tools from all configured MCP servers (maximal functionality)
- **How it works**: JSONRPC proxy that filters and routes tool calls transparently
### 3. **Sparse Mode Optimization**
To minimize context usage, only 3 meta-tools are shown initially:
- `mcp_discover` - Discover all available tools using JSONPath
## Core Architecture: Minimal Interface → Maximal Backend
### 1. **Sparse Mode Frontend** (What AI Sees)
Only 3 meta-tools are exposed, preventing context explosion:
- `mcp_discover` - Explore available tools without loading them into context
- `mcp_call` - Execute any tool by constructing JSON-RPC calls
- `onboarding` - Get/set identity-specific instructions
- `onboarding` - Identity-aware persistent instructions
### 2. **Transparent JSONRPC Proxy** (How It Works)
- **Intercepts** `tools/list` responses and replaces full catalogs with sparse tools
- **Routes** tool calls to appropriate internal or external MCP servers
- **Transforms** meta-tool calls into actual JSONRPC requests
- **Buffers** responses and handles async message routing
### 3. **Multi-Server Backend** (What's Available)
- **Built-in Servers**: Screen, Memory, Patterns, Onboarding (always available)
- **External Servers**: Any MCP server configured in `~/.claude/mcp-browser/config.yaml`
- **Runtime Discovery**: New servers added without restart via config monitoring
## Key Insight: Tool Discovery Without Context Pollution
Instead of loading hundreds of tool descriptions into context, AI can discover them on-demand:
```python
# Explore what's available (uses 0 additional context)
mcp_discover(jsonpath="$.tools[*].name")
# Get specific tool details only when needed
mcp_discover(jsonpath="$.tools[?(@.name=='brave_web_search')]")
# Execute discovered tools
mcp_call(method="tools/call", params={"name": "brave_web_search", "arguments": {...}})
```
## Discovery Examples
### Listing Available Servers
```python
# First, discover what MCP servers are available
mcp_discover(jsonpath="$.servers[*].name")
# Returns: ["builtin", "claude-code", "filesystem", etc.]
# Get detailed info about servers
mcp_discover(jsonpath="$.servers[*]")
```
### Basic Tool Discovery
```python
# Discover all available tools (built-in + external servers)
mcp_discover(jsonpath="$.tools[*].name")
@ -32,13 +64,20 @@ mcp_discover(jsonpath="$.tools[*].name")
# Get tools from specific server
mcp_discover(jsonpath="$.servers.brave-search.tools[*].name")
# Get all configured servers
mcp_discover(jsonpath="$.servers[*].name")
# Get tool details
mcp_discover(jsonpath="$.tools[?(@.name=='brave_web_search')]")
```
### Server-Specific Tool Discovery
```python
# Example: Get all Claude Code tools
mcp_discover(jsonpath="$.servers['claude-code'].tools[*].name")
# Returns: ["read_file", "write_file", "list_directory", etc.]
# Get details of a specific tool from claude-code server
mcp_discover(jsonpath="$.servers['claude-code'].tools[?(@.name=='read_file')]")
```
## Using External Server Tools
Once discovered, call any tool through `mcp_call`:
@ -61,6 +100,29 @@ mcp_call(
"arguments": {"query": "mcp-browser"}
}
)
# Example: Using Claude Code server to read a file
mcp_call(
method="tools/call",
params={
"name": "claude-code::read_file",
"arguments": {
"path": "/path/to/file.py",
"start_line": 1,
"end_line": 50
}
}
)
# Alternative syntax for server-namespaced tools
mcp_call(
method="tools/call",
params={
"name": "read_file",
"arguments": {"path": "/path/to/file.py"},
"server": "claude-code"
}
)
```
## Runtime Configuration

View File

@ -125,17 +125,21 @@ class OnboardingServer(BaseMCPServer):
content = self._format_onboarding(identity, data)
return self.content_text(content)
else:
# Try to load predefined markdown files first
predefined_file = Path(__file__).parent / f"{identity}.md"
if predefined_file.exists():
with open(predefined_file) as f:
predefined_content = f.read()
return self.content_text(predefined_content)
# Try to load default onboarding
default_file = self.onboarding_dir / "default.md"
default_file = Path(__file__).parent / "default.md"
if default_file.exists():
with open(default_file) as f:
default_content = f.read()
return self.content_text(
f"# Onboarding for {identity}\n\n"
f"No specific onboarding found. Using default:\n\n"
f"{default_content}"
)
return self.content_text(default_content)
else:
return self.content_text(
f"# Onboarding for {identity}\n\n"

View File

@ -0,0 +1,223 @@
# Xilope Production Environment - MCP Browser Guide
Welcome to Xilope's production MCP Browser setup! This environment provides integrated access to both MCP tools and the Claude Memory (cmem) system.
## Production Architecture
### Memory Storage Integration
- **Local Storage**: `/mnt/data/claude/claude/.mcp-memory/`
- **cmem Integration**: Bidirectional sync via `/mnt/data/claude/claude/bin/cmem`
- **Identity-Based**: Each project gets separate memory space
- **Persistent**: Memory survives across AI assistant sessions
### Built-in Servers Available
1. **Memory Server** (`builtin:memory::`):
- `task_add`, `task_list`, `task_update` - Task management with cmem sync
- `decision_add` - Decision tracking with reasoning
- `pattern_add`, `pattern_resolve` - Learning pattern management
- `knowledge_add`, `knowledge_get` - Fact storage and retrieval
- `project_switch` - Switch between project contexts
- `memory_summary` - Get overview of stored information
2. **Screen Server** (`builtin:screen::`):
- `create_session`, `execute`, `peek` - GNU screen management
- `list_sessions`, `kill_session` - Session lifecycle
- `enable_multiuser`, `attach_multiuser`, `add_user` - Collaboration
3. **Pattern Server** (`builtin:patterns::`):
- `add_pattern`, `list_patterns` - Auto-response pattern management
- `test_pattern`, `execute_pattern` - Pattern execution
4. **Onboarding Server** (`builtin:onboarding::`):
- `onboarding` - Identity-aware instructions
- `onboarding_list`, `onboarding_delete`, `onboarding_export` - Management
## Production Workflows
### Starting a Session
```python
# Check what servers are available
mcp_discover(jsonpath="$.servers[*].name")
# List all memory tools for task management
mcp_discover(jsonpath="$.tools[?(@.name =~ /memory|task/)]")
# Check current project context
mcp_call(
method="tools/call",
params={
"name": "builtin:memory::memory_summary",
"arguments": {}
}
)
```
### Task Management with cmem Sync
```python
# Add a new task (automatically syncs to cmem)
mcp_call(
method="tools/call",
params={
"name": "builtin:memory::task_add",
"arguments": {
"content": "Implement feature X",
"priority": "high",
"assignee": "next-ai"
}
}
)
# List active tasks
mcp_call(
method="tools/call",
params={
"name": "builtin:memory::task_list",
"arguments": {"status": "pending"}
}
)
# Update task status (syncs completion to cmem)
mcp_call(
method="tools/call",
params={
"name": "builtin:memory::task_update",
"arguments": {
"task_id": "abc123",
"status": "completed"
}
}
)
```
### Decision and Pattern Management
```python
# Record important decisions (synced to cmem)
mcp_call(
method="tools/call",
params={
"name": "builtin:memory::decision_add",
"arguments": {
"choice": "Use Docker for deployment",
"reasoning": "Simplifies environment management",
"alternatives": ["Native install", "VM deployment"]
}
}
)
# Add learning patterns (synced to cmem)
mcp_call(
method="tools/call",
params={
"name": "builtin:memory::pattern_add",
"arguments": {
"pattern": "Always test before commit",
"description": "Run full test suite before any git commit",
"priority": "high"
}
}
)
```
### Screen Session Management
```python
# Create a development session
mcp_call(
method="tools/call",
params={
"name": "builtin:screen::create_session",
"arguments": {
"session_name": "development",
"working_directory": "/mnt/data/claude/claude"
}
}
)
# Execute commands in session
mcp_call(
method="tools/call",
params={
"name": "builtin:screen::execute",
"arguments": {
"session_name": "development",
"command": "git status"
}
}
)
# Peek at session output
mcp_call(
method="tools/call",
params={
"name": "builtin:screen::peek",
"arguments": {"session_name": "development"}
}
)
```
### Project Context Switching
```python
# Switch to different project memory space
mcp_call(
method="tools/call",
params={
"name": "builtin:memory::project_switch",
"arguments": {"project_name": "mcp-browser"}
}
)
# Use onboarding for project-specific instructions
mcp_call(
method="tools/call",
params={
"name": "builtin:onboarding::onboarding",
"arguments": {
"identity": "mcp-browser",
"instructions": "Focus on context optimization and AI-first development"
}
}
)
```
## cmem Integration Details
### Memory Storage Structure
```
/mnt/data/claude/claude/.mcp-memory/
├── default/ # Default project space
│ ├── tasks.json # Task storage
│ ├── decisions.json # Decision history
│ ├── patterns.json # Learning patterns
│ └── knowledge.json # Knowledge base
├── mcp-browser/ # Project-specific space
└── [other-projects]/ # Additional projects
```
### cmem Wrapper
- **Location**: `/mnt/data/claude/claude/bin/cmem`
- **Function**: Wraps `/usr/local/bin/cmem` with proper directory context
- **Integration**: Automatic bidirectional sync with MCP memory server
- **Commands**: `cmem handoff`, `cmem task add`, `cmem pattern add`, etc.
### Sync Behavior
- **Automatic**: All memory operations sync to cmem in background
- **Graceful**: If cmem unavailable, operations continue locally
- **Identity-Aware**: Each project gets separate cmem context
- **Bidirectional**: Changes in either system propagate to the other
## Production Best Practices
1. **Always check memory summary** at start of session
2. **Use task management** to track work across AI sessions
3. **Record decisions** with reasoning for future reference
4. **Create patterns** to capture effective approaches
5. **Switch project contexts** when working on different codebases
6. **Use screen sessions** for persistent development environments
## Error Handling
If cmem sync fails:
- Operations continue with local storage
- Sync retries automatically when cmem becomes available
- No data loss occurs during temporary cmem unavailability
This production setup ensures seamless AI assistant transitions while maintaining full project context and memory across sessions.

View File

@ -115,6 +115,62 @@ class ScreenServer(BaseMCPServer):
"required": ["session"]
}
)
# Enable multiuser mode
self.register_tool(
name="enable_multiuser",
description="Enable multiuser mode for a screen session",
input_schema={
"type": "object",
"properties": {
"session": {
"type": "string",
"description": "Name of the screen session"
}
},
"required": ["session"]
}
)
# Attach to multiuser session
self.register_tool(
name="attach_multiuser",
description="Attach to a multiuser screen session (for external use)",
input_schema={
"type": "object",
"properties": {
"session": {
"type": "string",
"description": "Name of the screen session"
},
"user": {
"type": "string",
"description": "Optional username for access control"
}
},
"required": ["session"]
}
)
# Add user to multiuser session
self.register_tool(
name="add_user",
description="Add a user to a multiuser screen session",
input_schema={
"type": "object",
"properties": {
"session": {
"type": "string",
"description": "Name of the screen session"
},
"user": {
"type": "string",
"description": "Username to add"
}
},
"required": ["session", "user"]
}
)
async def handle_tool_call(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Handle screen tool calls."""
@ -129,6 +185,12 @@ class ScreenServer(BaseMCPServer):
return await self._list_sessions()
elif tool_name == "kill_session":
return await self._kill_session(arguments)
elif tool_name == "enable_multiuser":
return await self._enable_multiuser(arguments)
elif tool_name == "attach_multiuser":
return await self._attach_multiuser(arguments)
elif tool_name == "add_user":
return await self._add_user(arguments)
else:
raise Exception(f"Unknown tool: {tool_name}")
@ -189,9 +251,22 @@ class ScreenServer(BaseMCPServer):
if result.returncode != 0:
return self.content_text(f"Failed to peek at session: {result.stderr}")
# Read the output
with open(tmp_path, 'r') as f:
content = f.read()
# Read the output with proper encoding handling
try:
with open(tmp_path, 'rb') as f:
raw_content = f.read()
# Try to decode with UTF-8, replacing invalid sequences
content = raw_content.decode('utf-8', errors='replace')
except Exception:
# Fallback to reading with latin-1 which accepts all bytes
with open(tmp_path, 'r', encoding='latin-1') as f:
content = f.read()
# Clean ANSI escape sequences
import re
ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
content = ansi_escape.sub('', content)
# Get last N lines
output_lines = content.strip().split('\n')
@ -246,6 +321,51 @@ class ScreenServer(BaseMCPServer):
else:
return self.content_text(f"Failed to kill session: {result.stderr}")
async def _enable_multiuser(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Enable multiuser mode for a screen session."""
session = args["session"]
# Enable multiuser mode
cmd = ["screen", "-S", session, "-X", "multiuser", "on"]
result = await self._run_command(cmd)
if result.returncode == 0:
return self.content_text(f"Enabled multiuser mode for session '{session}'")
else:
return self.content_text(f"Failed to enable multiuser mode: {result.stderr}")
async def _attach_multiuser(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Provide instructions for attaching to a multiuser session."""
session = args["session"]
user = args.get("user", "")
# Check if session exists and is multiuser
check_result = await self._run_command(["screen", "-ls", session])
if session not in check_result.stdout:
return self.content_text(f"Session '{session}' not found")
# Provide attach command
if user:
attach_cmd = f"screen -x {user}/{session}"
else:
attach_cmd = f"screen -x {session}"
return self.content_text(f"To attach to multiuser session '{session}', run: {attach_cmd}")
async def _add_user(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Add a user to a multiuser screen session."""
session = args["session"]
user = args["user"]
# Add user to session access control list
cmd = ["screen", "-S", session, "-X", "acladd", user]
result = await self._run_command(cmd)
if result.returncode == 0:
return self.content_text(f"Added user '{user}' to session '{session}'")
else:
return self.content_text(f"Failed to add user: {result.stderr}")
async def _run_command(self, cmd: List[str]) -> subprocess.CompletedProcess:
"""Run a command and return the result."""
return await asyncio.to_thread(

View File

@ -0,0 +1,324 @@
#!/usr/bin/env python3
"""
Tmux MCP Server - tmux session management.
Provides tools for creating and managing persistent tmux sessions,
useful for long-running processes and maintaining shell state.
"""
import os
import sys
import asyncio
import subprocess
import json
from typing import Dict, Any, List, Optional
from pathlib import Path
# Add parent directory to path
sys.path.append(str(Path(__file__).parent.parent))
from base import BaseMCPServer
class TmuxServer(BaseMCPServer):
"""MCP server for tmux management."""
def __init__(self):
super().__init__("tmux-server", "1.0.0")
self._register_tools()
def _register_tools(self):
"""Register all tmux management tools."""
# Create session tool
self.register_tool(
name="create_session",
description="Create a new tmux session with optional initial command",
input_schema={
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Name for the tmux session"
},
"command": {
"type": "string",
"description": "Optional command to run in the session"
}
},
"required": ["name"]
}
)
# Execute command tool
self.register_tool(
name="execute",
description="Execute a command in an existing tmux session",
input_schema={
"type": "object",
"properties": {
"session": {
"type": "string",
"description": "Name of the tmux session"
},
"command": {
"type": "string",
"description": "Command to execute"
}
},
"required": ["session", "command"]
}
)
# Peek at session output
self.register_tool(
name="peek",
description="Get recent output from a tmux session (last 50 lines by default)",
input_schema={
"type": "object",
"properties": {
"session": {
"type": "string",
"description": "Name of the tmux session"
},
"lines": {
"type": "integer",
"description": "Number of lines to retrieve (default: 50)",
"default": 50
}
},
"required": ["session"]
}
)
# List sessions
self.register_tool(
name="list_sessions",
description="List all active tmux sessions",
input_schema={
"type": "object",
"properties": {}
}
)
# Kill session
self.register_tool(
name="kill_session",
description="Terminate a tmux session",
input_schema={
"type": "object",
"properties": {
"session": {
"type": "string",
"description": "Name of the tmux session to kill"
}
},
"required": ["session"]
}
)
# Attach to session (provides instructions)
self.register_tool(
name="attach_session",
description="Provide instructions for attaching to a tmux session",
input_schema={
"type": "object",
"properties": {
"session": {
"type": "string",
"description": "Name of the tmux session"
}
},
"required": ["session"]
}
)
# Share session (tmux supports multiple clients by default)
self.register_tool(
name="share_session",
description="Get instructions for sharing a tmux session with other users",
input_schema={
"type": "object",
"properties": {
"session": {
"type": "string",
"description": "Name of the tmux session"
}
},
"required": ["session"]
}
)
async def handle_tool_call(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Handle tmux tool calls."""
if tool_name == "create_session":
return await self._create_session(arguments)
elif tool_name == "execute":
return await self._execute_command(arguments)
elif tool_name == "peek":
return await self._peek_session(arguments)
elif tool_name == "list_sessions":
return await self._list_sessions()
elif tool_name == "kill_session":
return await self._kill_session(arguments)
elif tool_name == "attach_session":
return await self._attach_session(arguments)
elif tool_name == "share_session":
return await self._share_session(arguments)
else:
raise Exception(f"Unknown tool: {tool_name}")
async def _create_session(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Create a new tmux session."""
name = args["name"]
command = args.get("command")
# Check if session already exists
check_result = await self._run_command(["tmux", "list-sessions", "-F", "#{session_name}"])
if check_result.returncode == 0 and name in check_result.stdout.split('\n'):
return self.content_text(f"Session '{name}' already exists")
# Create session
cmd = ["tmux", "new-session", "-d", "-s", name]
if command:
cmd.append(command)
result = await self._run_command(cmd)
if result.returncode == 0:
return self.content_text(f"Created tmux session '{name}'" +
(f" running '{command}'" if command else ""))
else:
return self.content_text(f"Failed to create session: {result.stderr}")
async def _execute_command(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Execute a command in a tmux session."""
session = args["session"]
command = args["command"]
# Send command to tmux session
cmd = ["tmux", "send-keys", "-t", session, command, "Enter"]
result = await self._run_command(cmd)
if result.returncode == 0:
return self.content_text(f"Executed command in session '{session}'")
else:
return self.content_text(f"Failed to execute command: {result.stderr}")
async def _peek_session(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Get recent output from a tmux session."""
session = args["session"]
lines = args.get("lines", 50)
# Get pane content from tmux
cmd = ["tmux", "capture-pane", "-t", session, "-p"]
result = await self._run_command(cmd)
if result.returncode != 0:
return self.content_text(f"Failed to peek at session: {result.stderr}")
# Clean ANSI escape sequences
import re
ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
content = ansi_escape.sub('', result.stdout)
# Get last N lines
output_lines = content.strip().split('\n')
if len(output_lines) > lines:
output_lines = output_lines[-lines:]
output = '\n'.join(output_lines)
return self.content_text(output if output else "(No output)")
async def _list_sessions(self) -> Dict[str, Any]:
"""List all active tmux sessions."""
result = await self._run_command(["tmux", "list-sessions", "-F", "#{session_name}: #{?session_attached,attached,not attached} (#{session_windows} windows)"])
if result.returncode != 0:
if "no server running" in result.stderr.lower():
return self.content_text("No tmux server running (no active sessions)")
else:
return self.content_text(f"Error listing sessions: {result.stderr}")
if result.stdout.strip():
output = "Active tmux sessions:\n" + result.stdout.strip()
else:
output = "No active tmux sessions"
return self.content_text(output)
async def _kill_session(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Kill a tmux session."""
session = args["session"]
cmd = ["tmux", "kill-session", "-t", session]
result = await self._run_command(cmd)
if result.returncode == 0:
return self.content_text(f"Killed tmux session '{session}'")
else:
return self.content_text(f"Failed to kill session: {result.stderr}")
async def _attach_session(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Provide instructions for attaching to a tmux session."""
session = args["session"]
# Check if session exists
check_result = await self._run_command(["tmux", "list-sessions", "-F", "#{session_name}"])
if check_result.returncode != 0 or session not in check_result.stdout.split('\n'):
return self.content_text(f"Session '{session}' not found")
# Provide attach command
attach_cmd = f"tmux attach-session -t {session}"
return self.content_text(f"To attach to session '{session}', run: {attach_cmd}")
async def _share_session(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Get instructions for sharing a tmux session."""
session = args["session"]
# Check if session exists
check_result = await self._run_command(["tmux", "list-sessions", "-F", "#{session_name}"])
if check_result.returncode != 0 or session not in check_result.stdout.split('\n'):
return self.content_text(f"Session '{session}' not found")
instructions = f"""To share tmux session '{session}':
1. Multiple users can attach simultaneously:
tmux attach-session -t {session}
2. For read-only access:
tmux attach-session -t {session} -r
3. To create a new session attached to the same windows:
tmux new-session -t {session}
Note: tmux supports multiple clients by default - no special setup needed!"""
return self.content_text(instructions)
async def _run_command(self, cmd: List[str]) -> subprocess.CompletedProcess:
"""Run a command and return the result."""
return await asyncio.to_thread(
subprocess.run,
cmd,
capture_output=True,
text=True
)
if __name__ == "__main__":
# Check if tmux is installed
try:
subprocess.run(["tmux", "-V"], capture_output=True, check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
print("Error: tmux is not installed", file=sys.stderr)
print("Install it with: sudo apt-get install tmux", file=sys.stderr)
sys.exit(1)
# Run the server
server = TmuxServer()
asyncio.run(server.run())

136
setup.py
View File

@ -158,6 +158,137 @@ class TestCommand(Command):
print("✅ All tests passed!")
class GenerateApiDocs(Command):
"""Generate comprehensive MCP API documentation for AI consumption."""
description = 'Generate JSON API documentation of all MCP servers and tools'
user_options = []
def initialize_options(self):
pass
def finalize_options(self):
pass
def run(self):
"""Generate comprehensive MCP API documentation."""
print("Generating MCP API Documentation...")
import asyncio
from datetime import datetime
async def generate_docs():
# Import here to avoid circular dependencies
from mcp_browser import MCPBrowser
from pathlib import Path
try:
from pathlib import Path
import os
# Initialize MCP Browser with config from standard location
config_path = Path.home() / ".claude" / "mcp-browser" / "config.yaml"
print(f"Loading config from: {config_path}")
if not config_path.exists():
print(f"⚠ Config file not found, creating default")
config_path.parent.mkdir(parents=True, exist_ok=True)
# Initialize with config and built-in servers
browser = MCPBrowser(
config_path=config_path if config_path.exists() else None,
enable_builtin_servers=True
)
print("Waiting for server initialization...")
# Wait for initialization of servers
await asyncio.sleep(3)
# Perform tool discovery using the browser's call method
print("Discovering tools from all servers...")
try:
tools_response = await browser.call({
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list"
})
print(f"Tools response: {len(tools_response.get('result', {}).get('tools', []))} tools found")
except Exception as e:
print(f"Warning: Could not discover tools: {e}")
print(f"Registry has {len(browser.registry.raw_tool_list)} tools")
# Set server metadata for documentation
if hasattr(browser, 'config') and browser.config:
server_metadata = {
"servers": {}
}
for name, config in browser.config.servers.items():
server_metadata["servers"][name] = {
"command": getattr(config, 'command', []),
"description": getattr(config, 'description', ''),
"status": "configured",
"env": getattr(config, 'env', {}),
"tools": [] # Will be populated by get_full_api_documentation
}
browser.registry.set_metadata(server_metadata)
# Get comprehensive API documentation
api_doc = browser.registry.get_full_api_documentation()
api_doc["generation_timestamp"] = datetime.now().isoformat()
# Add runtime server information
if hasattr(browser, 'multi_server') and browser.multi_server:
server_status = {}
for name, server in browser.multi_server.servers.items():
server_status[name] = {
"status": "active" if hasattr(server, 'process') and server.process else "inactive",
"pid": getattr(server.process, 'pid', None) if hasattr(server, 'process') and server.process else None
}
api_doc["runtime_status"] = server_status
# Write to file
output_file = Path("mcp_api_documentation.json")
with open(output_file, 'w') as f:
import json
json.dump(api_doc, f, indent=2)
print(f"✓ Generated comprehensive MCP API documentation")
print(f"✓ Output: {output_file.absolute()}")
print(f"✓ Total servers: {api_doc['total_servers']}")
print(f"✓ Total tools: {api_doc['total_tools']}")
# Also print summary for immediate use
print("\n" + "=" * 60)
print("QUICK REFERENCE:")
print("=" * 60)
builtin = api_doc.get("builtin", {})
print(f"Built-in tools ({builtin.get('tool_count', 0)}):")
for tool in builtin.get("tools", [])[:5]: # Show first 5
print(f" - {tool.get('name', 'Unknown')}")
if builtin.get('tool_count', 0) > 5:
print(f" ... and {builtin.get('tool_count', 0) - 5} more")
print(f"\nExternal servers ({len(api_doc.get('servers', {}))}):")
for server_name, server_info in api_doc.get("servers", {}).items():
print(f" - {server_name}: {server_info.get('tool_count', 0)} tools")
print(f"\nDiscovery patterns available in documentation:")
for pattern_name, pattern in api_doc.get("discovery_patterns", {}).items():
print(f" - {pattern_name}: {pattern}")
# Clean up
await browser.close()
except Exception as e:
print(f"✗ Failed to generate API documentation: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
asyncio.run(generate_docs())
# Read long description
with open("README.md", "r", encoding="utf-8") as fh:
long_description = fh.read()
@ -165,7 +296,7 @@ with open("README.md", "r", encoding="utf-8") as fh:
setup(
name="mcp-browser",
version="0.1.0",
version="0.2.0",
description="A generic MCP browser with context optimization for AI systems",
long_description=long_description,
long_description_content_type="text/markdown",
@ -213,12 +344,13 @@ setup(
cmdclass={
'aidocs': GenerateAIDocs,
'test': TestCommand,
'gen_apidoc': GenerateApiDocs,
},
license="GPL-3.0-or-later",
classifiers=[
"Development Status :: 3 - Alpha",
"Intended Audience :: Developers",
"Topic :: Software Development :: Libraries :: Application Frameworks",
"License :: OSI Approved :: GNU General Public License v3 or later (GPLv3+)",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",

172
test_screen_multiuser.py Normal file
View File

@ -0,0 +1,172 @@
#!/usr/bin/env python3
"""Test multiuser functionality in screen server."""
import asyncio
import json
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mcp_browser import MCPBrowser
async def test_screen_multiuser():
"""Test screen multiuser session functionality."""
browser = MCPBrowser(enable_builtin_servers=True)
await browser.initialize()
print("=== Testing Screen Multiuser Functionality ===\n")
# Create a test session
print("1. Creating test session...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "create_session",
"arguments": {
"name": "multiuser-test",
"command": "bash"
}
}
})
print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}")
# Enable multiuser mode
print("\n2. Enabling multiuser mode...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "enable_multiuser",
"arguments": {
"session": "multiuser-test"
}
}
})
print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}")
# Add a user to the session
print("\n3. Adding user to session...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "add_user",
"arguments": {
"session": "multiuser-test",
"user": "testuser"
}
}
})
print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}")
# Get attach instructions
print("\n4. Getting attach instructions...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 4,
"method": "tools/call",
"params": {
"name": "attach_multiuser",
"arguments": {
"session": "multiuser-test",
"user": "testuser"
}
}
})
print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}")
# Execute a command in the multiuser session
print("\n5. Executing command in multiuser session...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 5,
"method": "tools/call",
"params": {
"name": "execute",
"arguments": {
"session": "multiuser-test",
"command": "echo 'Multiuser session test - Hello World!'"
}
}
})
print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}")
# Wait a moment for output
await asyncio.sleep(0.5)
# Test peek functionality with multiuser session
print("\n6. Testing peek with multiuser session...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 6,
"method": "tools/call",
"params": {
"name": "peek",
"arguments": {
"session": "multiuser-test",
"lines": 10
}
}
})
if "result" in response:
output = response["result"]["content"][0]["text"]
print(f" Output:\n{output}")
# Check if we can see the executed command output
if "Multiuser session test" in output:
print("\n✓ Peek works correctly with multiuser session!")
else:
print("\n⚠ Peek output might not show recent commands")
else:
print(f" Error: {response}")
# List sessions to see multiuser status
print("\n7. Listing sessions...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 7,
"method": "tools/call",
"params": {
"name": "list_sessions",
"arguments": {}
}
})
if "result" in response:
sessions_output = response["result"]["content"][0]["text"]
print(f" Sessions:\n{sessions_output}")
# Check if multiuser session is listed
if "multiuser-test" in sessions_output:
print("\n✓ Multiuser session is listed!")
else:
print("\n⚠ Session not found in list")
else:
print(f" Error: {response}")
# Clean up
print("\n8. Cleaning up...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 8,
"method": "tools/call",
"params": {
"name": "kill_session",
"arguments": {"session": "multiuser-test"}
}
})
print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}")
await browser.close()
if __name__ == "__main__":
asyncio.run(test_screen_multiuser())

101
test_screen_utf8.py Normal file
View File

@ -0,0 +1,101 @@
#!/usr/bin/env python3
"""Test UTF-8 handling in screen peek functionality."""
import asyncio
import json
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mcp_browser import MCPBrowser
async def test_screen_utf8():
"""Test screen peek with non-UTF8 content."""
browser = MCPBrowser(enable_builtin_servers=True)
await browser.initialize()
print("=== Testing Screen UTF-8 Handling ===\n")
# Create a test session
print("1. Creating test session...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "create_session",
"arguments": {
"name": "utf8-test",
"command": "bash"
}
}
})
print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}")
# Execute command that produces mixed encoding
print("\n2. Executing command with mixed encoding...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "execute",
"arguments": {
"session": "utf8-test",
"command": "echo -e 'UTF-8: café\\nBinary: \\x80\\x81\\x82\\nEmoji: 🤖'"
}
}
})
print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}")
# Wait a moment for output
await asyncio.sleep(0.5)
# Peek at the session
print("\n3. Peeking at session output...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "peek",
"arguments": {
"session": "utf8-test",
"lines": 10
}
}
})
if "result" in response:
output = response["result"]["content"][0]["text"]
print(f" Output:\n{output}")
# Check if we handled the encoding properly
if "café" in output and "🤖" in output:
print("\n✓ UTF-8 encoding handled correctly!")
else:
print("\n⚠ Some UTF-8 characters may not have been decoded properly")
else:
print(f" Error: {response}")
# Clean up
print("\n4. Cleaning up...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 4,
"method": "tools/call",
"params": {
"name": "kill_session",
"arguments": {"session": "utf8-test"}
}
})
print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}")
await browser.close()
if __name__ == "__main__":
asyncio.run(test_screen_utf8())

155
test_tmux_session.py Normal file
View File

@ -0,0 +1,155 @@
#!/usr/bin/env python3
"""Test tmux session functionality."""
import asyncio
import json
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mcp_browser import MCPBrowser
async def test_tmux_session():
"""Test tmux session functionality."""
browser = MCPBrowser(enable_builtin_servers=True)
await browser.initialize()
print("=== Testing Tmux Session Functionality ===\n")
# Create a test session
print("1. Creating test session...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "builtin:tmux::create_session",
"arguments": {
"name": "tmux-test",
"command": "bash"
}
}
})
print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}")
# Execute a command in the session
print("\n2. Executing command in session...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "builtin:tmux::execute",
"arguments": {
"session": "tmux-test",
"command": "echo 'Tmux session test - Hello World!'"
}
}
})
print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}")
# Wait a moment for output
await asyncio.sleep(0.5)
# Test peek functionality
print("\n3. Testing peek functionality...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "builtin:tmux::peek",
"arguments": {
"session": "tmux-test",
"lines": 10
}
}
})
if "result" in response:
output = response["result"]["content"][0]["text"]
print(f" Output:\n{output}")
# Check if we can see the executed command output
if "Tmux session test" in output:
print("\n✓ Peek works correctly!")
else:
print("\n⚠ Peek output might not show recent commands")
else:
print(f" Error: {response}")
# List sessions
print("\n4. Listing sessions...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 4,
"method": "tools/call",
"params": {
"name": "builtin:tmux::list_sessions",
"arguments": {}
}
})
if "result" in response:
sessions_output = response["result"]["content"][0]["text"]
print(f" Sessions:\n{sessions_output}")
# Check if session is listed
if "tmux-test" in sessions_output:
print("\n✓ Tmux session is listed!")
else:
print("\n⚠ Session not found in list")
else:
print(f" Error: {response}")
# Get attach instructions
print("\n5. Getting attach instructions...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 5,
"method": "tools/call",
"params": {
"name": "builtin:tmux::attach_session",
"arguments": {
"session": "tmux-test"
}
}
})
print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}")
# Test share instructions
print("\n6. Getting share instructions...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 6,
"method": "tools/call",
"params": {
"name": "builtin:tmux::share_session",
"arguments": {
"session": "tmux-test"
}
}
})
print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}")
# Clean up
print("\n7. Cleaning up...")
response = await browser.call({
"jsonrpc": "2.0",
"id": 7,
"method": "tools/call",
"params": {
"name": "builtin:tmux::kill_session",
"arguments": {"session": "tmux-test"}
}
})
print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}")
await browser.close()
if __name__ == "__main__":
asyncio.run(test_tmux_session())

View File

@ -54,7 +54,7 @@ class TestToolRegistry:
assert sparse[0]["name"] == "mcp_discover"
assert sparse[1]["name"] == "mcp_call"
assert sparse[2]["name"] == "onboarding"
assert "2 tools available" in sparse[0]["description"]
assert "2 hidden tools" in sparse[0]["description"]
class TestMessageFilter:

View File

@ -0,0 +1,312 @@
#!/usr/bin/env python3
"""
Test suite for cmem integration in memory server.
"""
import pytest
import asyncio
import json
import tempfile
from unittest.mock import Mock, AsyncMock, patch
from pathlib import Path
import sys
# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))
from mcp_servers.memory.memory_server import MemoryServer
class TestCmemIntegration:
"""Test cmem integration functionality."""
def setup_method(self):
"""Setup test environment with temporary directory."""
self.temp_dir = tempfile.mkdtemp()
def test_memory_server_initialization_default(self):
"""Test memory server initializes with default identity."""
with patch('pathlib.Path.home', return_value=Path(self.temp_dir)):
server = MemoryServer()
assert server.current_project == "default"
def test_memory_server_initialization_custom_identity(self):
"""Test memory server initializes with custom identity."""
with patch('pathlib.Path.home', return_value=Path(self.temp_dir)):
server = MemoryServer(identity="test_project")
assert server.current_project == "test_project"
def test_setup_cmem_integration_no_cmem(self):
"""Test cmem integration setup when cmem is not available."""
with patch('pathlib.Path.home', return_value=Path(self.temp_dir)):
with patch('subprocess.run') as mock_run:
mock_run.return_value.returncode = 1 # cmem not available
server = MemoryServer()
assert server.cmem_integration is False
def test_setup_cmem_integration_available(self):
"""Test cmem integration setup when cmem is available."""
with patch('pathlib.Path.home', return_value=Path(self.temp_dir)):
with patch('subprocess.run') as mock_run:
mock_run.return_value.returncode = 0 # cmem available
# Create mock .claude directory
claude_dir = Path(self.temp_dir) / ".claude"
claude_dir.mkdir()
sessions_dir = claude_dir / "sessions" / "test_session"
sessions_dir.mkdir(parents=True)
server = MemoryServer()
# Should have attempted cmem integration
assert hasattr(server, 'cmem_integration')
def test_create_cmem_bridges(self):
"""Test creation of cmem bridge files."""
with patch('pathlib.Path.home', return_value=Path(self.temp_dir)):
server = MemoryServer()
identity_dir = Path(self.temp_dir) / ".mcp-memory" / "test"
identity_dir.mkdir(parents=True)
session_dir = Path(self.temp_dir) / "session"
session_dir.mkdir()
server._create_cmem_bridges(identity_dir, session_dir)
bridge_dir = identity_dir / "cmem_bridge"
assert bridge_dir.exists()
info_file = bridge_dir / "info.json"
assert info_file.exists()
with open(info_file) as f:
bridge_info = json.load(f)
assert bridge_info["session_dir"] == str(session_dir)
assert bridge_info["integration_active"] is True
assert "last_sync" in bridge_info
@pytest.mark.asyncio
async def test_sync_task_to_cmem_add(self):
"""Test syncing task addition to cmem."""
with patch('pathlib.Path.home', return_value=Path(self.temp_dir)):
server = MemoryServer()
server.cmem_integration = True
from mcp_servers.memory.memory_server import Task
task = Task(
id="test-id",
content="Test task",
priority="high",
assignee="test_user"
)
with patch('asyncio.create_subprocess_exec') as mock_subprocess:
mock_process = AsyncMock()
mock_process.communicate.return_value = (b"", b"")
mock_subprocess.return_value = mock_process
await server._sync_task_to_cmem(task, "add")
# Verify subprocess was called with correct arguments
mock_subprocess.assert_called_once()
args = mock_subprocess.call_args[0]
assert args[0] == "cmem"
assert args[1] == "task"
assert args[2] == "add"
assert args[3] == "Test task"
assert "--priority" in args
assert "high" in args
assert "--assignee" in args
assert "test_user" in args
@pytest.mark.asyncio
async def test_sync_task_to_cmem_complete(self):
"""Test syncing task completion to cmem."""
with patch('pathlib.Path.home', return_value=Path(self.temp_dir)):
server = MemoryServer()
server.cmem_integration = True
from mcp_servers.memory.memory_server import Task
task = Task(
id="test-id",
content="Test task completion",
status="completed"
)
with patch('asyncio.create_subprocess_exec') as mock_subprocess:
mock_process = AsyncMock()
mock_process.communicate.return_value = (b"", b"")
mock_subprocess.return_value = mock_process
await server._sync_task_to_cmem(task, "complete")
# Verify subprocess was called for completion
mock_subprocess.assert_called_once()
args = mock_subprocess.call_args[0]
assert args[0] == "cmem"
assert args[1] == "task"
assert args[2] == "complete"
assert "Test task completion"[:50] in args[3] # Truncated content
@pytest.mark.asyncio
async def test_sync_pattern_to_cmem(self):
"""Test syncing pattern to cmem."""
with patch('pathlib.Path.home', return_value=Path(self.temp_dir)):
server = MemoryServer()
server.cmem_integration = True
from mcp_servers.memory.memory_server import Pattern
pattern = Pattern(
id="test-pattern-id",
pattern="Test pattern",
description="Pattern description",
priority="high"
)
with patch('asyncio.create_subprocess_exec') as mock_subprocess:
mock_process = AsyncMock()
mock_process.communicate.return_value = (b"", b"")
mock_subprocess.return_value = mock_process
await server._sync_pattern_to_cmem(pattern, "add")
# Verify subprocess was called with correct arguments
mock_subprocess.assert_called_once()
args = mock_subprocess.call_args[0]
assert args[0] == "cmem"
assert args[1] == "pattern"
assert args[2] == "add"
assert args[3] == "Test pattern"
assert args[4] == "Pattern description"
assert "--priority" in args
assert "high" in args
@pytest.mark.asyncio
async def test_sync_decision_to_cmem(self):
"""Test syncing decision to cmem."""
with patch('pathlib.Path.home', return_value=Path(self.temp_dir)):
server = MemoryServer()
server.cmem_integration = True
from mcp_servers.memory.memory_server import Decision
decision = Decision(
id="test-decision-id",
choice="Test choice",
reasoning="Test reasoning",
alternatives=["Alt 1", "Alt 2"]
)
with patch('asyncio.create_subprocess_exec') as mock_subprocess:
mock_process = AsyncMock()
mock_process.communicate.return_value = (b"", b"")
mock_subprocess.return_value = mock_process
await server._sync_decision_to_cmem(decision)
# Verify subprocess was called with correct arguments
mock_subprocess.assert_called_once()
args = mock_subprocess.call_args[0]
assert args[0] == "cmem"
assert args[1] == "decision"
assert args[2] == "Test choice"
assert args[3] == "Test reasoning"
assert args[4] == "Alt 1, Alt 2"
@pytest.mark.asyncio
async def test_sync_with_integration_disabled(self):
"""Test that sync methods do nothing when integration is disabled."""
with patch('pathlib.Path.home', return_value=Path(self.temp_dir)):
server = MemoryServer()
server.cmem_integration = False
from mcp_servers.memory.memory_server import Task
task = Task(id="test", content="test")
with patch('asyncio.create_subprocess_exec') as mock_subprocess:
await server._sync_task_to_cmem(task, "add")
# Should not have called subprocess
mock_subprocess.assert_not_called()
@pytest.mark.asyncio
async def test_sync_error_handling(self):
"""Test that sync errors are handled gracefully."""
with patch('pathlib.Path.home', return_value=Path(self.temp_dir)):
server = MemoryServer()
server.cmem_integration = True
from mcp_servers.memory.memory_server import Task
task = Task(id="test", content="test")
with patch('asyncio.create_subprocess_exec') as mock_subprocess:
mock_subprocess.side_effect = Exception("Subprocess error")
# Should not raise exception
await server._sync_task_to_cmem(task, "add")
@pytest.mark.asyncio
async def test_task_add_with_sync(self):
"""Test task addition triggers cmem sync."""
with patch('pathlib.Path.home', return_value=Path(self.temp_dir)):
server = MemoryServer()
server.cmem_integration = True
with patch.object(server, '_sync_task_to_cmem') as mock_sync:
mock_sync.return_value = None # Async function
result = await server._task_add({
"content": "Test task",
"priority": "high"
})
# Verify sync was called
mock_sync.assert_called_once()
args = mock_sync.call_args[0]
assert args[1] == "add" # action
assert args[0].content == "Test task"
# Verify task was added
assert "Added task:" in result["content"][0]["text"]
@pytest.mark.asyncio
async def test_task_update_completion_with_sync(self):
"""Test task completion triggers cmem sync."""
with patch('pathlib.Path.home', return_value=Path(self.temp_dir)):
server = MemoryServer()
server.cmem_integration = True
# Add a task first
task_id = "test-task-id"
server.tasks[task_id] = {
"id": task_id,
"content": "Test task",
"status": "pending",
"priority": "medium",
"assignee": None,
"created_at": "2025-01-01T00:00:00",
"completed_at": None
}
with patch.object(server, '_sync_task_to_cmem') as mock_sync:
mock_sync.return_value = None # Async function
result = await server._task_update({
"task_id": task_id,
"status": "completed"
})
# Verify sync was called
mock_sync.assert_called_once()
args = mock_sync.call_args[0]
assert args[1] == "complete" # action
# Verify task was updated
assert server.tasks[task_id]["status"] == "completed"
assert server.tasks[task_id]["completed_at"] is not None
if __name__ == "__main__":
pytest.main([__file__, "-v"])

View File

@ -0,0 +1,380 @@
#!/usr/bin/env python3
"""
Test suite for the enhanced interactive MCP client.
"""
import pytest
import asyncio
import json
from unittest.mock import Mock, AsyncMock, patch
from pathlib import Path
import sys
# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))
from mcp_browser.interactive_client import InteractiveMCPClient
from mcp_browser.proxy import MCPBrowser
class TestInteractiveMCPClient:
"""Test the interactive MCP client functionality."""
def setup_method(self):
"""Setup test environment."""
self.client = InteractiveMCPClient(use_daemon=False)
def test_initialization(self):
"""Test client initialization."""
assert self.client.server_name is None
assert self.client.use_daemon is False
assert self.client.tool_cache == {}
assert self.client.command_history == []
def test_completer_commands(self):
"""Test tab completion for commands."""
# Mock readline state
with patch('readline.get_line_buffer', return_value='help'):
matches = []
state = 0
while True:
match = self.client._completer('hel', state)
if match is None:
break
matches.append(match)
state += 1
assert 'help' in matches
def test_completer_tools(self):
"""Test tab completion includes tool names when cached."""
# Setup tool cache
self.client.tool_cache = {
'Bash': {'name': 'Bash', 'description': 'Execute bash commands'},
'mcp_discover': {'name': 'mcp_discover', 'description': 'Discover tools'}
}
with patch('readline.get_line_buffer', return_value='Bash'):
matches = []
state = 0
while True:
match = self.client._completer('Ba', state)
if match is None:
break
matches.append(match)
state += 1
assert 'Bash' in matches
@pytest.mark.asyncio
async def test_refresh_tools(self):
"""Test tool cache refresh functionality."""
# Mock MCP browser
mock_browser = AsyncMock()
mock_browser.call.return_value = {
"result": {
"tools": [
{"name": "test_tool", "description": "Test tool"},
{"name": "another_tool", "description": "Another test tool"}
]
}
}
self.client.browser = mock_browser
await self.client._refresh_tools()
assert len(self.client.tool_cache) == 2
assert "test_tool" in self.client.tool_cache
assert "another_tool" in self.client.tool_cache
@pytest.mark.asyncio
async def test_call_mcp_browser(self):
"""Test MCP call through browser."""
mock_browser = AsyncMock()
expected_response = {"result": {"test": "data"}}
mock_browser.call.return_value = expected_response
self.client.browser = mock_browser
request = {"jsonrpc": "2.0", "id": 1, "method": "test"}
response = await self.client._call_mcp(request)
assert response == expected_response
mock_browser.call.assert_called_once_with(request)
@pytest.mark.asyncio
async def test_call_mcp_client(self):
"""Test MCP call through daemon client."""
mock_client = AsyncMock()
expected_response = {"result": {"test": "data"}}
mock_client.call.return_value = expected_response
self.client.client = mock_client
self.client.browser = None
request = {"jsonrpc": "2.0", "id": 1, "method": "test"}
response = await self.client._call_mcp(request)
assert response == expected_response
mock_client.call.assert_called_once_with(request)
def test_generate_sample_args_string(self):
"""Test sample argument generation for string properties."""
schema = {
"properties": {
"query": {"type": "string"},
"jsonpath": {"type": "string"},
"command": {"type": "string"}
}
}
args = self.client._generate_sample_args(schema)
assert args["jsonpath"] == "$.tools[*].name"
assert args["query"] == "test query"
assert args["command"] == "sample_command"
def test_generate_sample_args_types(self):
"""Test sample argument generation for different types."""
schema = {
"properties": {
"text": {"type": "string"},
"enabled": {"type": "boolean"},
"count": {"type": "number"},
"items": {"type": "array"},
"config": {"type": "object"}
}
}
args = self.client._generate_sample_args(schema)
assert isinstance(args["text"], str)
assert isinstance(args["enabled"], bool)
assert isinstance(args["count"], (int, float))
assert isinstance(args["items"], list)
assert isinstance(args["config"], dict)
def test_generate_sample_args_examples(self):
"""Test sample argument generation uses examples when available."""
schema = {
"properties": {
"query": {
"type": "string",
"example": "example query"
}
}
}
args = self.client._generate_sample_args(schema)
assert args["query"] == "example query"
@pytest.mark.asyncio
async def test_execute_tool_call(self):
"""Test tool execution with proper result display."""
mock_browser = AsyncMock()
mock_browser.call.return_value = {
"result": {
"content": [
{"type": "text", "text": "Test result"}
]
}
}
self.client.browser = mock_browser
# Capture output
with patch('builtins.print') as mock_print:
await self.client._execute_tool_call("test_tool", {"arg": "value"})
# Verify MCP call was made
mock_browser.call.assert_called_once()
call_args = mock_browser.call.call_args[0][0]
assert call_args["method"] == "tools/call"
assert call_args["params"]["name"] == "test_tool"
assert call_args["params"]["arguments"] == {"arg": "value"}
# Verify output was printed
mock_print.assert_called()
@pytest.mark.asyncio
async def test_execute_tool_call_error(self):
"""Test tool execution error handling."""
mock_browser = AsyncMock()
mock_browser.call.return_value = {
"error": {
"code": -32603,
"message": "Tool execution failed"
}
}
self.client.browser = mock_browser
with patch('builtins.print') as mock_print:
await self.client._execute_tool_call("test_tool", {})
# Check that error was printed
print_calls = [str(call) for call in mock_print.call_args_list]
assert any("Error:" in call for call in print_calls)
def test_display_result_text_content(self):
"""Test display of text content results."""
result = {
"content": [
{"type": "text", "text": "Hello, World!"}
]
}
with patch('builtins.print') as mock_print:
self.client._display_result(result)
# Verify text was printed
calls = [call[0][0] for call in mock_print.call_args_list]
assert "Hello, World!" in calls
def test_display_result_image_content(self):
"""Test display of image content results."""
result = {
"content": [
{"type": "image", "url": "http://example.com/image.png"}
]
}
with patch('builtins.print') as mock_print:
self.client._display_result(result)
# Verify image info was printed
calls = [call[0][0] for call in mock_print.call_args_list]
assert any("Image:" in call for call in calls)
def test_display_result_raw_data(self):
"""Test display of raw result data."""
result = {"key": "value", "number": 42}
with patch('builtins.print') as mock_print:
self.client._display_result(result)
# Verify JSON was printed
calls = [call[0][0] for call in mock_print.call_args_list]
assert any("Result:" in call for call in calls)
@pytest.mark.asyncio
async def test_execute_command_help(self):
"""Test help command execution."""
with patch('builtins.print') as mock_print:
await self.client._execute_command("help")
# Verify help was printed
mock_print.assert_called()
print_calls = [str(call) for call in mock_print.call_args_list]
assert any("MCP Browser Interactive Commands" in call for call in print_calls)
@pytest.mark.asyncio
async def test_execute_command_list(self):
"""Test list command execution."""
# Setup tool cache
self.client.tool_cache = {
'test_tool': {'name': 'test_tool', 'description': 'A test tool'},
'bash_tool': {'name': 'bash_tool', 'description': 'Bash execution tool'}
}
with patch('builtins.print') as mock_print:
await self.client._execute_command("list bash")
# Verify filtered tools were printed
print_calls = [str(call) for call in mock_print.call_args_list]
assert any("bash_tool" in call for call in print_calls)
@pytest.mark.asyncio
async def test_execute_command_refresh(self):
"""Test refresh command execution."""
mock_browser = AsyncMock()
mock_browser.call.return_value = {
"result": {"tools": [{"name": "new_tool", "description": "New tool"}]}
}
self.client.browser = mock_browser
with patch('builtins.print') as mock_print:
await self.client._execute_command("refresh")
# Verify tool cache was updated
assert "new_tool" in self.client.tool_cache
# Verify refresh message was printed
print_calls = [str(call) for call in mock_print.call_args_list]
assert any("Tool cache refreshed" in call for call in print_calls)
@pytest.mark.asyncio
async def test_execute_command_unknown_tool(self):
"""Test handling of unknown direct tool calls."""
with patch('builtins.print') as mock_print:
await self.client._execute_command("unknown_tool arg1")
# Verify error message was printed
print_calls = [str(call) for call in mock_print.call_args_list]
assert any("Unknown tool:" in call for call in print_calls)
class TestInteractiveMCPClientIntegration:
"""Integration tests for interactive client."""
@pytest.mark.asyncio
async def test_full_workflow_mock(self):
"""Test a complete workflow with mocked dependencies."""
client = InteractiveMCPClient(use_daemon=False)
# Mock browser
mock_browser = AsyncMock()
# Mock tools/list response
tools_response = {
"result": {
"tools": [
{
"name": "mcp_discover",
"description": "Discover tools",
"inputSchema": {
"type": "object",
"properties": {
"jsonpath": {"type": "string"}
},
"required": ["jsonpath"]
}
}
]
}
}
# Mock discover response
discover_response = {
"result": {
"content": [
{"type": "text", "text": '["mcp_discover", "mcp_call", "onboarding"]'}
]
}
}
# Configure mock to return different responses based on call
def mock_call(request):
if request.get("method") == "tools/list":
return tools_response
elif (request.get("method") == "tools/call" and
request.get("params", {}).get("name") == "mcp_discover"):
return discover_response
else:
return {"error": {"code": -32601, "message": "Method not found"}}
mock_browser.call.side_effect = mock_call
client.browser = mock_browser
# Test tool refresh
await client._refresh_tools()
assert "mcp_discover" in client.tool_cache
# Test discovery command
with patch('builtins.print'):
await client._execute_command("discover $.tools[*].name")
# Verify calls were made
assert mock_browser.call.call_count >= 2
if __name__ == "__main__":
pytest.main([__file__, "-v"])

View File

@ -49,7 +49,7 @@ def test_sparse_mode():
assert sparse[2]["name"] == "onboarding"
# Check tool count in description
assert "5 tools available" in sparse[0]["description"]
assert "5 hidden tools" in sparse[0]["description"]
print("✓ Sparse mode tests passed")