ClaudeandClaude Fable 5
36da3f7024
Adds files_mcp/tools/requests.py with 9 Pydantic body models (one per tool, since fastapi-mcp passes args as JSON body) and files_mcp/tools/files.py with 9 thin wrappers that call path_guard then fs_ops and return model_dump() form. Adds tests/unit/test_files_tool.py covering happy path, sandbox escape, and KeyError/FileExistsError propagation for each wrapper. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
77 lines
2.6 KiB
Python
77 lines
2.6 KiB
Python
# coding=utf-8
|
|
"""
|
|
@Time :2026/6/30
|
|
@Author :tao.chen
|
|
|
|
Business wrappers: each tool function logs on entry, calls path_guard
|
|
on every user-supplied path, delegates to fs_ops, and returns the
|
|
Pydantic-serialized result. No business logic of its own — same shape
|
|
as spark_executor/tools/connections.py.
|
|
"""
|
|
from common.logging import logger
|
|
from files_mcp.core import fs_ops, path_guard
|
|
|
|
|
|
def create_file(*, path: str, content: str, overwrite: bool = False) -> dict:
|
|
logger.debug(
|
|
f"create_file enter path={path} content_len={len(content)} overwrite={overwrite}"
|
|
)
|
|
p = path_guard.resolve_and_check(path)
|
|
return fs_ops.create(p, content, overwrite).model_dump()
|
|
|
|
|
|
def read_file(*, path: str) -> dict:
|
|
logger.debug(f"read_file enter path={path}")
|
|
p = path_guard.resolve_and_check(path)
|
|
return fs_ops.read(p)
|
|
|
|
|
|
def update_file(*, path: str, content: str) -> dict:
|
|
logger.debug(f"update_file enter path={path} content_len={len(content)}")
|
|
p = path_guard.resolve_and_check(path)
|
|
return fs_ops.update(p, content).model_dump()
|
|
|
|
|
|
def delete_file(*, path: str, recursive: bool = False) -> dict:
|
|
logger.debug(f"delete_file enter path={path} recursive={recursive}")
|
|
p = path_guard.resolve_and_check(path)
|
|
return fs_ops.delete(p, recursive).model_dump()
|
|
|
|
|
|
def list_dir(
|
|
*, path: str, recursive: bool = False, max_depth: int = 1
|
|
) -> dict:
|
|
logger.debug(
|
|
f"list_dir enter path={path} recursive={recursive} max_depth={max_depth}"
|
|
)
|
|
p = path_guard.resolve_and_check(path)
|
|
return {"path": str(p), "entries": [e.model_dump() for e in fs_ops.list_dir(p, recursive, max_depth)]}
|
|
|
|
|
|
def stat(*, path: str) -> dict:
|
|
logger.debug(f"stat enter path={path}")
|
|
p = path_guard.resolve_and_check(path)
|
|
return fs_ops.stat(p).model_dump()
|
|
|
|
|
|
def search(*, path: str, pattern: str, recursive: bool = True) -> dict:
|
|
logger.debug(
|
|
f"search enter path={path} pattern={pattern!r} recursive={recursive}"
|
|
)
|
|
p = path_guard.resolve_and_check(path)
|
|
return {"pattern": pattern, "matches": fs_ops.search(p, pattern, recursive)}
|
|
|
|
|
|
def move(*, src: str, dst: str, overwrite: bool = False) -> dict:
|
|
logger.debug(f"move enter src={src} dst={dst} overwrite={overwrite}")
|
|
s = path_guard.resolve_and_check(src)
|
|
d = path_guard.resolve_and_check(dst)
|
|
return fs_ops.move(s, d, overwrite).model_dump()
|
|
|
|
|
|
def copy(*, src: str, dst: str, overwrite: bool = False) -> dict:
|
|
logger.debug(f"copy enter src={src} dst={dst} overwrite={overwrite}")
|
|
s = path_guard.resolve_and_check(src)
|
|
d = path_guard.resolve_and_check(dst)
|
|
return fs_ops.copy(s, d, overwrite).model_dump()
|