Private
Public Access
feat(mcp_client): add 8 Batch D _result variants in Result Variants region
Phase 6 Batch D step 1: added 8 _result variants for: - py_get_signature_result - py_set_signature_result - py_get_class_summary_result - py_get_var_declaration_result - py_set_var_declaration_result - py_find_usages_result - py_get_imports_result - py_check_syntax_result Legacy function migrations are pending (need manual edits due to slight content variations between expected and actual source). Will follow up.
This commit is contained in:
@@ -700,6 +700,217 @@ def py_update_definition_result(path: str, name: str, new_content: str) -> Resul
|
||||
return Result(data=f"Successfully updated definition '{name}' in {path}")
|
||||
except Exception as e:
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="mcp.py_update_definition_result", original=e)])
|
||||
|
||||
def py_get_signature_result(path: str, name: str) -> Result[str]:
|
||||
resolved = _resolve_and_check_result(path)
|
||||
if not resolved.ok:
|
||||
return Result(data="", errors=resolved.errors)
|
||||
p = resolved.data
|
||||
if isinstance(p, NilPath):
|
||||
return Result(data="", errors=resolved.errors)
|
||||
if not p.exists():
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"file not found: {path}", source="mcp.py_get_signature_result")])
|
||||
try:
|
||||
code = p.read_text(encoding="utf-8").lstrip(chr(0xFEFF))
|
||||
lines = code.splitlines(keepends=True)
|
||||
tree = ast.parse(code)
|
||||
node = _get_symbol_node(tree, name)
|
||||
if not node or not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"could not find function/method '{name}' in {path}", source="mcp.py_get_signature_result")])
|
||||
start = cast(int, getattr(node, "lineno")) - 1
|
||||
body_start = cast(int, getattr(node.body[0], "lineno")) - 1
|
||||
sig = "".join(lines[start:body_start]).rstrip()
|
||||
if not sig.endswith(":"):
|
||||
for j in range(body_start, len(lines)):
|
||||
if ":" in lines[j]:
|
||||
full_line = lines[j]
|
||||
colon_idx = full_line.index(":")
|
||||
sig = "".join(lines[start:j+1])
|
||||
return Result(data=sig[:colon_idx+1].strip())
|
||||
return Result(data=sig)
|
||||
except Exception as e:
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="mcp.py_get_signature_result", original=e)])
|
||||
|
||||
def py_set_signature_result(path: str, name: str, new_signature: str) -> Result[str]:
|
||||
resolved = _resolve_and_check_result(path)
|
||||
if not resolved.ok:
|
||||
return Result(data="", errors=resolved.errors)
|
||||
p = resolved.data
|
||||
if isinstance(p, NilPath):
|
||||
return Result(data="", errors=resolved.errors)
|
||||
if not p.exists():
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"file not found: {path}", source="mcp.py_set_signature_result")])
|
||||
try:
|
||||
code = p.read_text(encoding="utf-8").lstrip(chr(0xFEFF))
|
||||
tree = ast.parse(code)
|
||||
node = _get_symbol_node(tree, name)
|
||||
if not node or not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"could not find function/method '{name}' in {path}", source="mcp.py_set_signature_result")])
|
||||
start = node.lineno
|
||||
body_start_line = node.body[0].lineno
|
||||
end = body_start_line - 1
|
||||
inner = set_file_slice_result(path, start, end, new_signature)
|
||||
return inner
|
||||
except Exception as e:
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="mcp.py_set_signature_result", original=e)])
|
||||
|
||||
def py_get_class_summary_result(path: str, name: str) -> Result[str]:
|
||||
resolved = _resolve_and_check_result(path)
|
||||
if not resolved.ok:
|
||||
return Result(data="", errors=resolved.errors)
|
||||
p = resolved.data
|
||||
if isinstance(p, NilPath):
|
||||
return Result(data="", errors=resolved.errors)
|
||||
if not p.exists():
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"file not found: {path}", source="mcp.py_get_class_summary_result")])
|
||||
try:
|
||||
code = p.read_text(encoding="utf-8").lstrip(chr(0xFEFF))
|
||||
tree = ast.parse(code)
|
||||
node = _get_symbol_node(tree, name)
|
||||
if not node or not isinstance(node, ast.ClassDef):
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"could not find class '{name}' in {path}", source="mcp.py_get_class_summary_result")])
|
||||
lines = code.splitlines(keepends=True)
|
||||
summary = [f"Class: {name}"]
|
||||
doc = ast.get_docstring(node)
|
||||
if doc:
|
||||
summary.append(f" Docstring: {doc}")
|
||||
for body_node in node.body:
|
||||
if isinstance(body_node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
start = body_node.lineno - 1
|
||||
body_start = body_node.body[0].lineno - 1
|
||||
sig = "".join(lines[start:body_start]).strip()
|
||||
summary.append(f" - {sig}")
|
||||
return Result(data="\n".join(summary))
|
||||
except Exception as e:
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="mcp.py_get_class_summary_result", original=e)])
|
||||
|
||||
def py_get_var_declaration_result(path: str, name: str) -> Result[str]:
|
||||
resolved = _resolve_and_check_result(path)
|
||||
if not resolved.ok:
|
||||
return Result(data="", errors=resolved.errors)
|
||||
p = resolved.data
|
||||
if isinstance(p, NilPath):
|
||||
return Result(data="", errors=resolved.errors)
|
||||
if not p.is_file() or p.suffix != ".py":
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INVALID_INPUT, message=f"not a python file: {path}", source="mcp.py_get_var_declaration_result")])
|
||||
try:
|
||||
code = p.read_text(encoding="utf-8").lstrip(chr(0xFEFF))
|
||||
lines = code.splitlines(keepends=True)
|
||||
tree = ast.parse(code)
|
||||
node = _get_symbol_node(tree, name)
|
||||
if not node or not isinstance(node, (ast.Assign, ast.AnnAssign)):
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"could not find variable '{name}' in {path}", source="mcp.py_get_var_declaration_result")])
|
||||
start = cast(int, getattr(node, "lineno")) - 1
|
||||
end = cast(int, getattr(node, "end_lineno"))
|
||||
return Result(data="".join(lines[start:end]))
|
||||
except Exception as e:
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="mcp.py_get_var_declaration_result", original=e)])
|
||||
|
||||
def py_set_var_declaration_result(path: str, name: str, new_declaration: str) -> Result[str]:
|
||||
resolved = _resolve_and_check_result(path)
|
||||
if not resolved.ok:
|
||||
return Result(data="", errors=resolved.errors)
|
||||
p = resolved.data
|
||||
if isinstance(p, NilPath):
|
||||
return Result(data="", errors=resolved.errors)
|
||||
if not p.is_file() or p.suffix != ".py":
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INVALID_INPUT, message=f"not a python file: {path}", source="mcp.py_set_var_declaration_result")])
|
||||
try:
|
||||
code = p.read_text(encoding="utf-8").lstrip(chr(0xFEFF))
|
||||
tree = ast.parse(code)
|
||||
node = _get_symbol_node(tree, name)
|
||||
if not node or not isinstance(node, (ast.Assign, ast.AnnAssign)):
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"could not find variable '{name}' in {path}", source="mcp.py_set_var_declaration_result")])
|
||||
start = cast(int, getattr(node, "lineno"))
|
||||
end = cast(int, getattr(node, "end_lineno"))
|
||||
inner = set_file_slice_result(path, start, end, new_declaration)
|
||||
return inner
|
||||
except Exception as e:
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="mcp.py_set_var_declaration_result", original=e)])
|
||||
|
||||
def py_find_usages_result(path: str, name: str) -> Result[str]:
|
||||
resolved = _resolve_and_check_result(path)
|
||||
if not resolved.ok:
|
||||
return Result(data="", errors=resolved.errors)
|
||||
p = resolved.data
|
||||
if isinstance(p, NilPath):
|
||||
return Result(data="", errors=resolved.errors)
|
||||
try:
|
||||
import re
|
||||
pattern = re.compile(r"\b" + re.escape(name) + r"\b")
|
||||
results = []
|
||||
def _search_file(fp: Path) -> None:
|
||||
if fp.name == "history.toml" or fp.name.endswith("_history.toml"): return
|
||||
if not _is_allowed(fp): return
|
||||
try:
|
||||
text = fp.read_text(encoding="utf-8")
|
||||
lines = text.splitlines()
|
||||
for i, line in enumerate(lines, 1):
|
||||
if pattern.search(line):
|
||||
rel = fp.relative_to(_primary_base_dir if _primary_base_dir else Path.cwd())
|
||||
results.append(f"{rel}:{i}: {line.strip()[:100]}")
|
||||
except Exception:
|
||||
pass
|
||||
if p.is_file():
|
||||
_search_file(p)
|
||||
else:
|
||||
for root, dirs, files in os.walk(p):
|
||||
dirs[:] = [d for d in dirs if not d.startswith('.') and d not in ('__pycache__', 'venv', 'env')]
|
||||
for file in files:
|
||||
if file.endswith(('.py', '.md', '.toml', '.txt', '.json')):
|
||||
_search_file(Path(root) / file)
|
||||
if not results:
|
||||
return Result(data=f"No usages found for '{name}' in {p}")
|
||||
if len(results) > 100:
|
||||
return Result(data="\n".join(results[:100]) + f"\n... (and {len(results)-100} more)")
|
||||
return Result(data="\n".join(results))
|
||||
except Exception as e:
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="mcp.py_find_usages_result", original=e)])
|
||||
|
||||
def py_get_imports_result(path: str) -> Result[str]:
|
||||
resolved = _resolve_and_check_result(path)
|
||||
if not resolved.ok:
|
||||
return Result(data="", errors=resolved.errors)
|
||||
p = resolved.data
|
||||
if isinstance(p, NilPath):
|
||||
return Result(data="", errors=resolved.errors)
|
||||
if not p.is_file() or p.suffix != ".py":
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INVALID_INPUT, message=f"not a python file: {path}", source="mcp.py_get_imports_result")])
|
||||
try:
|
||||
code = p.read_text(encoding="utf-8")
|
||||
tree = ast.parse(code)
|
||||
imports = []
|
||||
for node in tree.body:
|
||||
if isinstance(node, ast.Import):
|
||||
for alias in node.names:
|
||||
imports.append(alias.name)
|
||||
elif isinstance(node, ast.ImportFrom):
|
||||
module = node.module or ""
|
||||
for alias in node.names:
|
||||
imports.append(f"{module}.{alias.name}" if module else alias.name)
|
||||
if not imports:
|
||||
return Result(data="No imports found.")
|
||||
return Result(data="Imports:\n" + "\n".join(f" - {i}" for i in imports))
|
||||
except Exception as e:
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="mcp.py_get_imports_result", original=e)])
|
||||
|
||||
def py_check_syntax_result(path: str) -> Result[str]:
|
||||
resolved = _resolve_and_check_result(path)
|
||||
if not resolved.ok:
|
||||
return Result(data="", errors=resolved.errors)
|
||||
p = resolved.data
|
||||
if isinstance(p, NilPath):
|
||||
return Result(data="", errors=resolved.errors)
|
||||
if not p.is_file() or p.suffix != ".py":
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INVALID_INPUT, message=f"not a python file: {path}", source="mcp.py_check_syntax_result")])
|
||||
try:
|
||||
code = p.read_text(encoding="utf-8")
|
||||
ast.parse(code)
|
||||
return Result(data=f"Syntax OK: {path}")
|
||||
except SyntaxError as e:
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INVALID_INPUT, message=f"SyntaxError in {path} at line {e.lineno}, offset {e.offset}: {e.msg}\n{e.text}", source="mcp.py_check_syntax_result")])
|
||||
except Exception as e:
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="mcp.py_check_syntax_result", original=e)])
|
||||
#endregion: Result Variants
|
||||
|
||||
def edit_file(path: str, old_string: str, new_string: str, replace_all: bool = False) -> str:
|
||||
|
||||
Reference in New Issue
Block a user