mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-27 04:20:09 +00:00
zooqueenandHanzo Dev
920f767417
Run ruff check --fix and ruff format across the repo to fix ~495 auto-fixable lint errors (mostly I001 import sorting) and reformat 225 files to match the py312 target. Co-authored-by: Hanzo Dev <[email protected]>
76 lines
2.1 KiB
Python
76 lines
2.1 KiB
Python
"""Write tool - write/create files."""
|
|
|
|
from typing import Optional, Annotated
|
|
from pathlib import Path
|
|
|
|
from pydantic import Field
|
|
from mcp.server import FastMCP
|
|
from hanzo_async import mkdir, write_file
|
|
from mcp.server.fastmcp import Context as MCPContext
|
|
|
|
from hanzo_tools.core import FileSystemTool, PermissionManager, auto_timeout
|
|
|
|
|
|
class WriteTool(FileSystemTool):
|
|
"""Write content to a file."""
|
|
|
|
name = "write"
|
|
|
|
@property
|
|
def description(self) -> str:
|
|
return """Write content to a file (creates or overwrites).
|
|
|
|
Args:
|
|
file_path: Absolute path to the file
|
|
content: Content to write
|
|
|
|
Returns:
|
|
Success message or error
|
|
"""
|
|
|
|
def __init__(self, permission_manager: Optional[PermissionManager] = None):
|
|
super().__init__(permission_manager)
|
|
|
|
@auto_timeout("write")
|
|
async def call(
|
|
self,
|
|
ctx: MCPContext,
|
|
file_path: str,
|
|
content: str,
|
|
**kwargs,
|
|
) -> str:
|
|
"""Write content to file."""
|
|
validation = self.validate_path(file_path)
|
|
if not validation:
|
|
return validation.error_message
|
|
|
|
if not self.is_path_allowed(file_path):
|
|
return f"Error: Access denied to path: {file_path}"
|
|
|
|
path = Path(file_path)
|
|
|
|
try:
|
|
# Create parent directories if needed
|
|
await mkdir(path.parent, parents=True, exist_ok=True)
|
|
|
|
# Write content
|
|
await write_file(path, content, encoding="utf-8")
|
|
|
|
return f"Successfully wrote {len(content)} bytes to {file_path}"
|
|
|
|
except Exception as e:
|
|
return f"Error writing file: {e}"
|
|
|
|
def register(self, mcp_server: FastMCP) -> None:
|
|
"""Register with MCP server."""
|
|
tool_instance = self
|
|
|
|
@mcp_server.tool()
|
|
async def write(
|
|
file_path: Annotated[str, Field(description="Absolute path to the file")],
|
|
content: Annotated[str, Field(description="Content to write")],
|
|
ctx: MCPContext = None,
|
|
) -> str:
|
|
"""Write content to a file."""
|
|
return await tool_instance.call(ctx, file_path=file_path, content=content)
|