Private
Public Access
0
0

more mcp org

This commit is contained in:
2026-06-07 01:28:01 -04:00
parent fd34467b55
commit 59d32ba96d
+202 -233
View File
@@ -241,10 +241,8 @@ def list_directory(path: str) -> str:
def search_files(path: str, pattern: 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) p, err = _resolve_and_check(path)
if err or p is None: 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: 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.
Return the heuristic summary for a file (same as the initial context block). For .toml: table keys. For .md: headings. Others: line count + preview.
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) p, err = _resolve_and_check(path)
if err or p is None: if err or p is None:
@@ -292,220 +288,6 @@ def get_file_summary(path: str) -> str:
except Exception as e: except Exception as e:
return f"ERROR summarising '{path}': {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: def get_file_slice(path: str, start_line: int, end_line: int) -> str:
"""Return a specific line range from a file.""" """Return a specific line range from a file."""
p, err = _resolve_and_check(path) 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: 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. Replace exact string match in a file. Preserves indentation and line endings.
Drop-in replacement for native edit tool that destroys 1-space indentation. 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: except Exception as e:
return f"ERROR: {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]: 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.""" """Helper to find an AST node by name (Class, Function, or Variable). Supports dot notation."""
parts = name.split(".") parts = name.split(".")
@@ -624,7 +563,44 @@ def _get_symbol_node(tree: ast.AST, name: str) -> Optional[ast.AST]:
current = found current = found
return current 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: 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): async def add_server(self, config: models.MCPServerConfig):
""" """
Add and start a new MCP server from a configuration object. 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] [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): async def stop_all(self):
""" """
Stop all managed MCP servers and clear the registry. 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] [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: def get_all_tools(self) -> dict:
""" """
Retrieve a dictionary of all tools available across all managed servers. 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] [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: async def async_dispatch(self, tool_name: str, tool_input: dict) -> str:
""" """
Dispatch a tool call to the appropriate external MCP server asynchronously. 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] [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: def get_external_mcp_manager() -> ExternalMCPManager:
""" """
Retrieve the global ExternalMCPManager instance. 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] [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}' return f'ERROR: unknown MCP tool {tool_name}'
def get_tool_schemas() -> list[dict[str, Any]]: 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] [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]