Skip to content

Commit fc8c930

Browse files
author
Yoshihiro Takahara
committed
refactor: move handlers to separate directory
- Create handlers directory - Extract base handler class - Move each handler to its own file - Update server.py to use new handlers - Fix newline handling in append handler
1 parent 79770fa commit fc8c930

10 files changed

+834
-702
lines changed

‎src/mcp_text_editor/handlers/__init__.py

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
"""Handlers for MCP Text Editor."""
2+
3+
from .append_text_file_contents import AppendTextFileContentsHandler
4+
from .create_text_file import CreateTextFileHandler
5+
from .delete_text_file_contents import DeleteTextFileContentsHandler
6+
from .edit_text_file_contents import EditTextFileContentsHandler
7+
from .get_text_file_contents import GetTextFileContentsHandler
8+
from .insert_text_file_contents import InsertTextFileContentsHandler
9+
from .patch_text_file_contents import PatchTextFileContentsHandler
10+
11+
__all__ = [
12+
"AppendTextFileContentsHandler",
13+
"CreateTextFileHandler",
14+
"DeleteTextFileContentsHandler",
15+
"EditTextFileContentsHandler",
16+
"GetTextFileContentsHandler",
17+
"InsertTextFileContentsHandler",
18+
"PatchTextFileContentsHandler",
19+
]

‎src/mcp_text_editor/handlers/append_text_file_contents.py

Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
"""Handler for appending content to text files."""
2+
3+
import json
4+
import logging
5+
import os
6+
import traceback
7+
from typing import Any, Dict, Sequence
8+
9+
from mcp.types import TextContent, Tool
10+
11+
from .base import BaseHandler
12+
13+
logger = logging.getLogger("mcp-text-editor")
14+
15+
16+
class AppendTextFileContentsHandler(BaseHandler):
17+
"""Handler for appending content to an existing text file."""
18+
19+
name = "append_text_file_contents"
20+
description = "Append content to an existing text file. The file must exist."
21+
22+
def get_tool_description(self) -> Tool:
23+
"""Get the tool description."""
24+
return Tool(
25+
name=self.name,
26+
description=self.description,
27+
inputSchema={
28+
"type": "object",
29+
"properties": {
30+
"path": {
31+
"type": "string",
32+
"description": "Path to the text file. File path must be absolute.",
33+
},
34+
"contents": {
35+
"type": "string",
36+
"description": "Content to append to the file",
37+
},
38+
"file_hash": {
39+
"type": "string",
40+
"description": "Hash of the file contents for concurrency control. it should be matched with the file_hash when get_text_file_contents is called.",
41+
},
42+
"encoding": {
43+
"type": "string",
44+
"description": "Text encoding (default: 'utf-8')",
45+
"default": "utf-8",
46+
},
47+
},
48+
"required": ["path", "contents", "file_hash"],
49+
},
50+
)
51+
52+
async def run_tool(self, arguments: Dict[str, Any]) -> Sequence[TextContent]:
53+
"""Execute the tool with given arguments."""
54+
try:
55+
if "path" not in arguments:
56+
raise RuntimeError("Missing required argument: path")
57+
if "contents" not in arguments:
58+
raise RuntimeError("Missing required argument: contents")
59+
if "file_hash" not in arguments:
60+
raise RuntimeError("Missing required argument: file_hash")
61+
62+
file_path = arguments["path"]
63+
if not os.path.isabs(file_path):
64+
raise RuntimeError(f"File path must be absolute: {file_path}")
65+
66+
# Check if file exists
67+
if not os.path.exists(file_path):
68+
raise RuntimeError(f"File does not exist: {file_path}")
69+
70+
encoding = arguments.get("encoding", "utf-8")
71+
72+
# Check file contents and hash before modification
73+
# Get file information and verify hash
74+
content, _, _, current_hash, total_lines, _ = (
75+
await self.editor.read_file_contents(file_path, encoding=encoding)
76+
)
77+
78+
# Verify file hash
79+
if current_hash != arguments["file_hash"]:
80+
raise RuntimeError("File hash mismatch - file may have been modified")
81+
82+
# Ensure the append content ends with newline
83+
append_content = arguments["contents"]
84+
if not append_content.endswith("\n"):
85+
append_content += "\n"
86+
87+
# Create patch for append operation
88+
result = await self.editor.edit_file_contents(
89+
file_path,
90+
expected_hash=arguments["file_hash"],
91+
patches=[
92+
{
93+
"start": total_lines + 1,
94+
"end": None,
95+
"contents": append_content,
96+
"range_hash": "",
97+
}
98+
],
99+
encoding=encoding,
100+
)
101+
102+
return [TextContent(type="text", text=json.dumps(result, indent=2))]
103+
104+
except Exception as e:
105+
logger.error(f"Error processing request: {str(e)}")
106+
logger.error(traceback.format_exc())
107+
raise RuntimeError(f"Error processing request: {str(e)}") from e

‎src/mcp_text_editor/handlers/base.py

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
"""Base handler for MCP Text Editor."""
2+
3+
from typing import Any, Dict, Sequence
4+
5+
from mcp.types import TextContent, Tool
6+
7+
from ..text_editor import TextEditor
8+
9+
10+
class BaseHandler:
11+
"""Base class for handlers."""
12+
13+
name: str = ""
14+
description: str = ""
15+
16+
def __init__(self):
17+
"""Initialize the handler."""
18+
self.editor = TextEditor()
19+
20+
def get_tool_description(self) -> Tool:
21+
"""Get the tool description."""
22+
raise NotImplementedError
23+
24+
async def run_tool(self, arguments: Dict[str, Any]) -> Sequence[TextContent]:
25+
"""Execute the tool with given arguments."""
26+
raise NotImplementedError

‎src/mcp_text_editor/handlers/create_text_file.py

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,87 @@
1+
"""Handler for creating new text files."""
2+
3+
import json
4+
import logging
5+
import os
6+
import traceback
7+
from typing import Any, Dict, Sequence
8+
9+
from mcp.types import TextContent, Tool
10+
11+
from .base import BaseHandler
12+
13+
logger = logging.getLogger("mcp-text-editor")
14+
15+
16+
class CreateTextFileHandler(BaseHandler):
17+
"""Handler for creating a new text file."""
18+
19+
name = "create_text_file"
20+
description = (
21+
"Create a new text file with given content. The file must not exist already."
22+
)
23+
24+
def get_tool_description(self) -> Tool:
25+
"""Get the tool description."""
26+
return Tool(
27+
name=self.name,
28+
description=self.description,
29+
inputSchema={
30+
"type": "object",
31+
"properties": {
32+
"path": {
33+
"type": "string",
34+
"description": "Path to the text file. File path must be absolute.",
35+
},
36+
"contents": {
37+
"type": "string",
38+
"description": "Content to write to the file",
39+
},
40+
"encoding": {
41+
"type": "string",
42+
"description": "Text encoding (default: 'utf-8')",
43+
"default": "utf-8",
44+
},
45+
},
46+
"required": ["path", "contents"],
47+
},
48+
)
49+
50+
async def run_tool(self, arguments: Dict[str, Any]) -> Sequence[TextContent]:
51+
"""Execute the tool with given arguments."""
52+
try:
53+
if "path" not in arguments:
54+
raise RuntimeError("Missing required argument: path")
55+
if "contents" not in arguments:
56+
raise RuntimeError("Missing required argument: contents")
57+
58+
file_path = arguments["path"]
59+
if not os.path.isabs(file_path):
60+
raise RuntimeError(f"File path must be absolute: {file_path}")
61+
62+
# Check if file already exists
63+
if os.path.exists(file_path):
64+
raise RuntimeError(f"File already exists: {file_path}")
65+
66+
encoding = arguments.get("encoding", "utf-8")
67+
68+
# Create new file using edit_file_contents with empty expected_hash
69+
result = await self.editor.edit_file_contents(
70+
file_path,
71+
expected_hash="", # Empty hash for new file
72+
patches=[
73+
{
74+
"start": 1,
75+
"end": None,
76+
"contents": arguments["contents"],
77+
"range_hash": "", # Empty range_hash for new file
78+
}
79+
],
80+
encoding=encoding,
81+
)
82+
return [TextContent(type="text", text=json.dumps(result, indent=2))]
83+
84+
except Exception as e:
85+
logger.error(f"Error processing request: {str(e)}")
86+
logger.error(traceback.format_exc())
87+
raise RuntimeError(f"Error processing request: {str(e)}") from e

‎src/mcp_text_editor/handlers/delete_text_file_contents.py

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
"""Handler for deleting content from text files."""
2+
3+
import json
4+
import logging
5+
import os
6+
import traceback
7+
from typing import Any, Dict, Sequence
8+
9+
from mcp.types import TextContent, Tool
10+
11+
from .base import BaseHandler
12+
13+
logger = logging.getLogger("mcp-text-editor")
14+
15+
16+
class DeleteTextFileContentsHandler(BaseHandler):
17+
"""Handler for deleting content from a text file."""
18+
19+
name = "delete_text_file_contents"
20+
description = "Delete specified content ranges from a text file. The file must exist. File paths must be absolute."
21+
22+
def get_tool_description(self) -> Tool:
23+
"""Get the tool description."""
24+
return Tool(
25+
name=self.name,
26+
description=self.description,
27+
inputSchema={
28+
"type": "object",
29+
"properties": {
30+
"file_path": {
31+
"type": "string",
32+
"description": "Path to the text file. File path must be absolute.",
33+
},
34+
"file_hash": {
35+
"type": "string",
36+
"description": "Hash of the file contents for concurrency control. it should be matched with the file_hash when get_text_file_contents is called.",
37+
},
38+
"ranges": {
39+
"type": "array",
40+
"description": "List of line ranges to delete",
41+
"items": {
42+
"type": "object",
43+
"properties": {
44+
"start": {
45+
"type": "integer",
46+
"description": "Starting line number (1-based)",
47+
},
48+
"end": {
49+
"type": ["integer", "null"],
50+
"description": "Ending line number (null for end of file)",
51+
},
52+
"range_hash": {
53+
"type": "string",
54+
"description": "Hash of the content being deleted. it should be matched with the range_hash when get_text_file_contents is called with the same range.",
55+
},
56+
},
57+
"required": ["start", "range_hash"],
58+
},
59+
},
60+
"encoding": {
61+
"type": "string",
62+
"description": "Text encoding (default: 'utf-8')",
63+
"default": "utf-8",
64+
},
65+
},
66+
"required": ["file_path", "file_hash", "ranges"],
67+
},
68+
)
69+
70+
async def run_tool(self, arguments: Dict[str, Any]) -> Sequence[TextContent]:
71+
"""Execute the tool with given arguments."""
72+
try:
73+
# Input validation
74+
if "file_path" not in arguments:
75+
raise RuntimeError("Missing required argument: file_path")
76+
if "file_hash" not in arguments:
77+
raise RuntimeError("Missing required argument: file_hash")
78+
if "ranges" not in arguments:
79+
raise RuntimeError("Missing required argument: ranges")
80+
81+
file_path = arguments["file_path"]
82+
if not os.path.isabs(file_path):
83+
raise RuntimeError(f"File path must be absolute: {file_path}")
84+
85+
# Check if file exists
86+
if not os.path.exists(file_path):
87+
raise RuntimeError(f"File does not exist: {file_path}")
88+
89+
encoding = arguments.get("encoding", "utf-8")
90+
91+
# Create patches for deletion (replacing content with empty string)
92+
patches = [
93+
{
94+
"start": r["start"],
95+
"end": r["end"],
96+
"contents": "",
97+
"range_hash": r["range_hash"],
98+
}
99+
for r in arguments["ranges"]
100+
]
101+
102+
# Use the existing edit_file_contents method
103+
result = await self.editor.edit_file_contents(
104+
file_path,
105+
expected_hash=arguments["file_hash"],
106+
patches=patches,
107+
encoding=encoding,
108+
)
109+
110+
return [TextContent(type="text", text=json.dumps(result, indent=2))]
111+
112+
except Exception as e:
113+
logger.error(f"Error processing request: {str(e)}")
114+
logger.error(traceback.format_exc())
115+
raise RuntimeError(f"Error processing request: {str(e)}") from e

0 commit comments

Comments
 (0)