From 59d32ba96d714bab0474ce224df18e3937f173fa Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sun, 7 Jun 2026 01:28:01 -0400 Subject: [PATCH] more mcp org --- src/mcp_client.py | 435 +++++++++++++++++++++------------------------- 1 file changed, 202 insertions(+), 233 deletions(-) diff --git a/src/mcp_client.py b/src/mcp_client.py index 144b1478..225c131d 100644 --- a/src/mcp_client.py +++ b/src/mcp_client.py @@ -241,10 +241,8 @@ def list_directory(path: str) -> str: def search_files(path: str, pattern: str) -> str: """ - - - Search for files matching a glob pattern within path. - pattern examples: '*.py', '**/*.toml', 'src/**/*.rs' + Search for files matching a glob pattern within path. + pattern examples: '*.py', '**/*.toml', 'src/**/*.rs' """ p, err = _resolve_and_check(path) if err or p is None: @@ -273,11 +271,9 @@ def search_files(path: str, pattern: str) -> str: def get_file_summary(path: str) -> str: """ - - - Return the heuristic summary for a file (same as the initial context block). - For .py files: imports, classes, methods, functions, constants. - For .toml: table keys. For .md: headings. Others: line count + preview. + Return the heuristic summary for a file (same as the initial context block). + For .py files: imports, classes, methods, functions, constants. + For .toml: table keys. For .md: headings. Others: line count + preview. """ p, err = _resolve_and_check(path) if err or p is None: @@ -292,220 +288,6 @@ def get_file_summary(path: str) -> str: except Exception as e: return f"ERROR summarising '{path}': {e}" -def py_get_skeleton(path: str) -> str: - """ - - - Returns a skeleton of a Python file (preserving docstrings, stripping function bodies). - """ - p, err = _resolve_and_check(path) - if err: - return err - assert p is not None - if not p.exists(): - return f"ERROR: file not found: {path}" - if not p.is_file() or p.suffix != ".py": - return f"ERROR: not a python file: {path}" - try: - from src.file_cache import ASTParser - code = p.read_text(encoding="utf-8") - parser = ASTParser("python") - return parser.get_skeleton(code) - except Exception as e: - return f"ERROR generating skeleton for '{path}': {e}" - -def ts_c_get_skeleton(path: str) -> str: - """ - - Returns a skeleton of a C file. - [C: tests/test_ts_c_tools.py:test_ts_c_get_skeleton] - """ - p, err = _resolve_and_check(path) - if err: return err - assert p is not None - if not p.exists(): return f"ERROR: file not found: {path}" - try: - from src.file_cache import ASTParser - code = p.read_text(encoding="utf-8") - parser = ASTParser("c") - return parser.get_skeleton(code, path=str(p)) - except Exception as e: - return f"ERROR generating skeleton for '{path}': {e}" - -def ts_cpp_get_skeleton(path: str) -> str: - """ - - Returns a skeleton of a C++ file. - [C: tests/test_gencpp_full_suite.py:test_gencpp_full_suite, tests/test_ts_cpp_tools.py:test_exhaustive_cpp_samples, tests/test_ts_cpp_tools.py:test_exhaustive_gencpp_samples, tests/test_ts_cpp_tools.py:test_ts_cpp_get_skeleton] - """ - p, err = _resolve_and_check(path) - if err: return err - assert p is not None - if not p.exists(): return f"ERROR: file not found: {path}" - try: - from src.file_cache import ASTParser - code = p.read_text(encoding="utf-8") - parser = ASTParser("cpp") - return parser.get_skeleton(code, path=str(p)) - except Exception as e: - return f"ERROR generating skeleton for '{path}': {e}" - -def py_get_code_outline(path: str) -> str: - """ - - - Returns a hierarchical outline of a code file (classes, functions, methods with line ranges). - """ - p, err = _resolve_and_check(path) - if err: - return err - assert p is not None - if not p.exists(): - return f"ERROR: file not found: {path}" - if not p.is_file(): - return f"ERROR: not a file: {path}" - try: - code = p.read_text(encoding="utf-8") - return outline_tool.get_outline(p, code) - except Exception as e: - return f"ERROR generating outline for '{path}': {e}" - -def ts_c_get_code_outline(path: str) -> str: - """ - - Returns a hierarchical outline of a C file. - [C: tests/test_ts_c_tools.py:test_ts_c_get_code_outline] - """ - p, err = _resolve_and_check(path) - if err: return err - assert p is not None - if not p.exists(): return f"ERROR: file not found: {path}" - try: - from src.file_cache import ASTParser - code = p.read_text(encoding="utf-8") - parser = ASTParser("c") - return parser.get_code_outline(code, path=str(p)) - except Exception as e: - return f"ERROR generating outline for '{path}': {e}" - -def ts_cpp_get_code_outline(path: str) -> str: - """ - - Returns a hierarchical outline of a C++ file. - [C: tests/test_gencpp_full_suite.py:test_gencpp_full_suite, tests/test_ts_cpp_tools.py:test_exhaustive_cpp_samples, tests/test_ts_cpp_tools.py:test_exhaustive_gencpp_samples, tests/test_ts_cpp_tools.py:test_ts_cpp_get_code_outline] - """ - p, err = _resolve_and_check(path) - if err: return err - assert p is not None - if not p.exists(): return f"ERROR: file not found: {path}" - try: - from src.file_cache import ASTParser - code = p.read_text(encoding="utf-8") - parser = ASTParser("cpp") - return parser.get_code_outline(code, path=str(p)) - except Exception as e: - return f"ERROR generating outline for '{path}': {e}" - -def ts_c_get_definition(path: str, name: str) -> str: - """Returns the source code for a specific definition in a C file.""" - p, err = _resolve_and_check(path) - if err: return err - assert p is not None - if not p.exists(): return f"ERROR: file not found: {path}" - try: - from src.file_cache import ASTParser - code = p.read_text(encoding="utf-8") - parser = ASTParser("c") - return parser.get_definition(code, name, path=str(p)) - except Exception as e: - return f"ERROR retrieving definition '{name}' from '{path}': {e}" - -def ts_cpp_get_definition(path: str, name: str) -> str: - """ - - Returns the source code for a specific definition in a C++ file. - [C: tests/test_ast_masking_core.py:test_ast_masking_gencpp_samples, tests/test_gencpp_full_suite.py:test_gencpp_full_suite, tests/test_ts_cpp_tools.py:test_exhaustive_cpp_samples, tests/test_ts_cpp_tools.py:test_exhaustive_gencpp_samples, tests/test_ts_cpp_tools.py:test_ts_cpp_update_definition, tests/test_ts_cpp_tools.py:test_ts_cpp_update_definition_gencpp] - """ - p, err = _resolve_and_check(path) - if err: return err - assert p is not None - if not p.exists(): return f"ERROR: file not found: {path}" - try: - from src.file_cache import ASTParser - code = p.read_text(encoding="utf-8") - parser = ASTParser("cpp") - return parser.get_definition(code, name, path=str(p)) - except Exception as e: - return f"ERROR retrieving definition '{name}' from '{path}': {e}" - -def ts_c_get_signature(path: str, name: str) -> str: - """Returns the signature part of a function in a C file.""" - p, err = _resolve_and_check(path) - if err: return err - assert p is not None - if not p.exists(): return f"ERROR: file not found: {path}" - try: - from src.file_cache import ASTParser - code = p.read_text(encoding="utf-8") - parser = ASTParser("c") - return parser.get_signature(code, name, path=str(p)) - except Exception as e: - return f"ERROR retrieving signature '{name}' from '{path}': {e}" - -def ts_cpp_get_signature(path: str, name: str) -> str: - """Returns the signature part of a function or method in a C++ file.""" - p, err = _resolve_and_check(path) - if err: return err - assert p is not None - if not p.exists(): return f"ERROR: file not found: {path}" - try: - from src.file_cache import ASTParser - code = p.read_text(encoding="utf-8") - parser = ASTParser("cpp") - return parser.get_signature(code, name, path=str(p)) - except Exception as e: - return f"ERROR retrieving signature '{name}' from '{path}': {e}" - -def ts_c_update_definition(path: str, name: str, new_content: str) -> str: - """Surgically replace the definition of a function in a C file.""" - p, err = _resolve_and_check(path) - if err: return err - assert p is not None - if not p.exists(): return f"ERROR: file not found: {path}" - try: - from src.file_cache import ASTParser - code = p.read_text(encoding="utf-8") - parser = ASTParser("c") - updated_code = parser.update_definition(code, name, new_content, path=str(p)) - if updated_code.startswith("ERROR:"): - return updated_code - p.write_text(updated_code, encoding="utf-8") - return f"Successfully updated definition '{name}' in {path}" - except Exception as e: - return f"ERROR updating definition '{name}' in '{path}': {e}" - -def ts_cpp_update_definition(path: str, name: str, new_content: str) -> str: - """ - - Surgically replace the definition of a class or function in a C++ file. - [C: tests/test_ts_cpp_tools.py:test_ts_cpp_update_definition, tests/test_ts_cpp_tools.py:test_ts_cpp_update_definition_gencpp] - """ - p, err = _resolve_and_check(path) - if err: return err - assert p is not None - if not p.exists(): return f"ERROR: file not found: {path}" - try: - from src.file_cache import ASTParser - code = p.read_text(encoding="utf-8") - parser = ASTParser("cpp") - updated_code = parser.update_definition(code, name, new_content, path=str(p)) - if updated_code.startswith("ERROR:"): - return updated_code - p.write_text(updated_code, encoding="utf-8") - return f"Successfully updated definition '{name}' in {path}" - except Exception as e: - return f"ERROR updating definition '{name}' in '{path}': {e}" - def get_file_slice(path: str, start_line: int, end_line: int) -> str: """Return a specific line range from a file.""" p, err = _resolve_and_check(path) @@ -545,8 +327,6 @@ def set_file_slice(path: str, start_line: int, end_line: int, new_content: str) def edit_file(path: str, old_string: str, new_string: str, replace_all: bool = False) -> str: """ - - Replace exact string match in a file. Preserves indentation and line endings. Drop-in replacement for native edit tool that destroys 1-space indentation. """ @@ -598,6 +378,165 @@ def get_git_diff(path: str, base_rev: str = "HEAD", head_rev: str = "") -> str: except Exception as e: return f"ERROR: {e}" +#region: C + +def ts_c_get_code_outline(path: str) -> str: + """ + Returns a hierarchical outline of a C file. + [C: tests/test_ts_c_tools.py:test_ts_c_get_code_outline] + """ + p, err = _resolve_and_check(path) + if err: return err + assert p is not None + if not p.exists(): return f"ERROR: file not found: {path}" + try: + from src.file_cache import ASTParser + code = p.read_text(encoding="utf-8") + parser = ASTParser("c") + return parser.get_code_outline(code, path=str(p)) + except Exception as e: + return f"ERROR generating outline for '{path}': {e}" + +def ts_c_get_definition(path: str, name: str) -> str: + """Returns the source code for a specific definition in a C file.""" + p, err = _resolve_and_check(path) + if err: return err + assert p is not None + if not p.exists(): return f"ERROR: file not found: {path}" + try: + from src.file_cache import ASTParser + code = p.read_text(encoding="utf-8") + parser = ASTParser("c") + return parser.get_definition(code, name, path=str(p)) + except Exception as e: + return f"ERROR retrieving definition '{name}' from '{path}': {e}" + +def ts_c_get_signature(path: str, name: str) -> str: + """Returns the signature part of a function in a C file.""" + p, err = _resolve_and_check(path) + if err: return err + assert p is not None + if not p.exists(): return f"ERROR: file not found: {path}" + try: + from src.file_cache import ASTParser + code = p.read_text(encoding="utf-8") + parser = ASTParser("c") + return parser.get_signature(code, name, path=str(p)) + except Exception as e: + return f"ERROR retrieving signature '{name}' from '{path}': {e}" + +def ts_c_update_definition(path: str, name: str, new_content: str) -> str: + """Surgically replace the definition of a function in a C file.""" + p, err = _resolve_and_check(path) + if err: return err + assert p is not None + if not p.exists(): return f"ERROR: file not found: {path}" + try: + from src.file_cache import ASTParser + code = p.read_text(encoding="utf-8") + parser = ASTParser("c") + updated_code = parser.update_definition(code, name, new_content, path=str(p)) + if updated_code.startswith("ERROR:"): + return updated_code + p.write_text(updated_code, encoding="utf-8") + return f"Successfully updated definition '{name}' in {path}" + except Exception as e: + return f"ERROR updating definition '{name}' in '{path}': {e}" + +#endregion: C + +#region: C++ + +def ts_cpp_get_skeleton(path: str) -> str: + """ + Returns a skeleton of a C++ file. + [C: tests/test_gencpp_full_suite.py:test_gencpp_full_suite, tests/test_ts_cpp_tools.py:test_exhaustive_cpp_samples, tests/test_ts_cpp_tools.py:test_exhaustive_gencpp_samples, tests/test_ts_cpp_tools.py:test_ts_cpp_get_skeleton] + """ + p, err = _resolve_and_check(path) + if err: return err + assert p is not None + if not p.exists(): return f"ERROR: file not found: {path}" + try: + from src.file_cache import ASTParser + code = p.read_text(encoding="utf-8") + parser = ASTParser("cpp") + return parser.get_skeleton(code, path=str(p)) + except Exception as e: + return f"ERROR generating skeleton for '{path}': {e}" + +def ts_cpp_get_code_outline(path: str) -> str: + """ + Returns a hierarchical outline of a C++ file. + [C: tests/test_gencpp_full_suite.py:test_gencpp_full_suite, tests/test_ts_cpp_tools.py:test_exhaustive_cpp_samples, tests/test_ts_cpp_tools.py:test_exhaustive_gencpp_samples, tests/test_ts_cpp_tools.py:test_ts_cpp_get_code_outline] + """ + p, err = _resolve_and_check(path) + if err: return err + assert p is not None + if not p.exists(): return f"ERROR: file not found: {path}" + try: + from src.file_cache import ASTParser + code = p.read_text(encoding="utf-8") + parser = ASTParser("cpp") + return parser.get_code_outline(code, path=str(p)) + except Exception as e: + return f"ERROR generating outline for '{path}': {e}" + +def ts_cpp_get_definition(path: str, name: str) -> str: + """ + Returns the source code for a specific definition in a C++ file. + [C: tests/test_ast_masking_core.py:test_ast_masking_gencpp_samples, tests/test_gencpp_full_suite.py:test_gencpp_full_suite, tests/test_ts_cpp_tools.py:test_exhaustive_cpp_samples, tests/test_ts_cpp_tools.py:test_exhaustive_gencpp_samples, tests/test_ts_cpp_tools.py:test_ts_cpp_update_definition, tests/test_ts_cpp_tools.py:test_ts_cpp_update_definition_gencpp] + """ + p, err = _resolve_and_check(path) + if err: return err + assert p is not None + if not p.exists(): return f"ERROR: file not found: {path}" + try: + from src.file_cache import ASTParser + code = p.read_text(encoding="utf-8") + parser = ASTParser("cpp") + return parser.get_definition(code, name, path=str(p)) + except Exception as e: + return f"ERROR retrieving definition '{name}' from '{path}': {e}" + +def ts_cpp_get_signature(path: str, name: str) -> str: + """Returns the signature part of a function or method in a C++ file.""" + p, err = _resolve_and_check(path) + if err: return err + assert p is not None + if not p.exists(): return f"ERROR: file not found: {path}" + try: + from src.file_cache import ASTParser + code = p.read_text(encoding="utf-8") + parser = ASTParser("cpp") + return parser.get_signature(code, name, path=str(p)) + except Exception as e: + return f"ERROR retrieving signature '{name}' from '{path}': {e}" + +def ts_cpp_update_definition(path: str, name: str, new_content: str) -> str: + """ + Surgically replace the definition of a class or function in a C++ file. + [C: tests/test_ts_cpp_tools.py:test_ts_cpp_update_definition, tests/test_ts_cpp_tools.py:test_ts_cpp_update_definition_gencpp] + """ + p, err = _resolve_and_check(path) + if err: return err + assert p is not None + if not p.exists(): return f"ERROR: file not found: {path}" + try: + from src.file_cache import ASTParser + code = p.read_text(encoding="utf-8") + parser = ASTParser("cpp") + updated_code = parser.update_definition(code, name, new_content, path=str(p)) + if updated_code.startswith("ERROR:"): + return updated_code + p.write_text(updated_code, encoding="utf-8") + return f"Successfully updated definition '{name}' in {path}" + except Exception as e: + return f"ERROR updating definition '{name}' in '{path}': {e}" + +#endregion: C++ + +#region: Python AST + def _get_symbol_node(tree: ast.AST, name: str) -> Optional[ast.AST]: """Helper to find an AST node by name (Class, Function, or Variable). Supports dot notation.""" parts = name.split(".") @@ -624,7 +563,44 @@ def _get_symbol_node(tree: ast.AST, name: str) -> Optional[ast.AST]: current = found return current -#region: Python AST +def py_get_skeleton(path: str) -> str: + """ + Returns a skeleton of a Python file (preserving docstrings, stripping function bodies). + """ + p, err = _resolve_and_check(path) + if err: + return err + assert p is not None + if not p.exists(): + return f"ERROR: file not found: {path}" + if not p.is_file() or p.suffix != ".py": + return f"ERROR: not a python file: {path}" + try: + from src.file_cache import ASTParser + code = p.read_text(encoding="utf-8") + parser = ASTParser("python") + return parser.get_skeleton(code) + except Exception as e: + return f"ERROR generating skeleton for '{path}': {e}" + +def py_get_code_outline(path: str) -> str: + """ + Returns a hierarchical outline of a code file (classes, functions, methods with line ranges). + """ + p, err = _resolve_and_check(path) + if err: + return err + assert p is not None + if not p.exists(): + return f"ERROR: file not found: {path}" + if not p.is_file(): + return f"ERROR: not a file: {path}" + try: + code = p.read_text(encoding="utf-8") + return outline_tool.get_outline(p, code) + except Exception as e: + return f"ERROR generating outline for '{path}': {e}" + def py_get_symbol_info(path: str, name: str) -> tuple[str, int] | str: """ @@ -1282,7 +1258,6 @@ class ExternalMCPManager: async def add_server(self, config: models.MCPServerConfig): """ - Add and start a new MCP server from a configuration object. [C: tests/test_external_mcp.py:test_external_mcp_real_process, tests/test_external_mcp.py:test_get_tool_schemas_includes_external] """ @@ -1295,7 +1270,6 @@ class ExternalMCPManager: async def stop_all(self): """ - Stop all managed MCP servers and clear the registry. [C: tests/test_external_mcp.py:test_external_mcp_real_process, tests/test_external_mcp.py:test_get_tool_schemas_includes_external, tests/test_external_mcp_e2e.py:test_external_mcp_e2e_refresh_and_call] """ @@ -1305,7 +1279,6 @@ class ExternalMCPManager: def get_all_tools(self) -> dict: """ - Retrieve a dictionary of all tools available across all managed servers. [C: tests/test_external_mcp.py:test_external_mcp_real_process, tests/test_external_mcp_e2e.py:test_external_mcp_e2e_refresh_and_call] """ @@ -1321,7 +1294,6 @@ class ExternalMCPManager: async def async_dispatch(self, tool_name: str, tool_input: dict) -> str: """ - Dispatch a tool call to the appropriate external MCP server asynchronously. [C: src/rag_engine.py:RAGEngine._async_search_mcp, tests/test_external_mcp.py:test_external_mcp_real_process] """ @@ -1334,7 +1306,6 @@ _external_mcp_manager = ExternalMCPManager() def get_external_mcp_manager() -> ExternalMCPManager: """ - Retrieve the global ExternalMCPManager instance. [C: tests/test_external_mcp.py:test_get_tool_schemas_includes_external, tests/test_external_mcp_e2e.py:test_external_mcp_e2e_refresh_and_call] """ @@ -1514,8 +1485,6 @@ async def async_dispatch(tool_name: str, tool_input: dict[str, Any]) -> str: return f'ERROR: unknown MCP tool {tool_name}' - - def get_tool_schemas() -> list[dict[str, Any]]: """ [C: tests/test_arch_boundary_phase2.py:TestArchBoundaryPhase2.test_mcp_client_dispatch_completeness, tests/test_external_mcp.py:test_get_tool_schemas_includes_external, tests/test_mcp_client_beads.py:test_bd_mcp_tools]