4
4
from mcp .server .fastmcp import FastMCP
5
5
from codegen import Codebase
6
6
7
+
7
8
@dataclass
8
9
class CodebaseState :
9
10
"""Class to manage codebase state and parsing."""
11
+
10
12
parse_task : Optional [asyncio .Task ] = None
11
13
parsed_codebase : Optional [Codebase ] = None
12
14
log_buffer : List [str ] = field (default_factory = list )
@@ -23,30 +25,32 @@ def reset(self) -> None:
23
25
self .parsed_codebase .reset ()
24
26
self .log_buffer .clear ()
25
27
28
+
26
29
# Initialize FastMCP server
27
30
mcp = FastMCP (
28
31
"codegen-mcp-server" ,
29
- instructions = """This server provides tools to parse and modify a codebase using codemods.
32
+ instructions = """This server provides tools to parse and modify a codebase using codemods.
30
33
It can initiate parsing, check parsing status, and execute codemods.""" ,
31
34
)
32
35
33
36
# Initialize state
34
37
state = CodebaseState ()
35
38
39
+
36
40
def capture_output (* args , ** kwargs ) -> None :
37
41
"""Capture and log output messages."""
38
42
for arg in args :
39
43
state .log_buffer .append (str (arg ))
40
44
45
+
41
46
@mcp .tool (name = "parse_codebase" , description = "Initiate codebase parsing" )
42
- async def parse_codebase (
43
- codebase_path : Annotated [str , "path to the codebase to be parsed" ]
44
- ) -> Dict [str , str ]:
47
+ async def parse_codebase (codebase_path : Annotated [str , "path to the codebase to be parsed" ]) -> Dict [str , str ]:
45
48
if not state .parse_task or state .parse_task .done ():
46
49
state .parse_task = asyncio .create_task (state .parse (codebase_path ))
47
50
return {"message" : "Codebase parsing initiated, this may take some time depending on the size of the codebase. Use the `check_parsing_status` tool to check if the parse has completed." }
48
51
return {"message" : "Codebase is already being parsed." }
49
52
53
+
50
54
@mcp .tool (name = "check_parsing_status" , description = "Check if codebase parsing has completed" )
51
55
async def check_parsing_status () -> Dict [str , str ]:
52
56
if not state .parse_task :
@@ -55,15 +59,14 @@ async def check_parsing_status() -> Dict[str, str]:
55
59
return {"message" : "Codebase parsing completed." }
56
60
return {"message" : "Codebase parsing in progress." }
57
61
62
+
58
63
@mcp .tool (name = "execute_codemod" , description = "Execute a codemod on the codebase" )
59
- async def execute_codemod (
60
- codemod : Annotated [str , "The python codemod code to execute on the codebase" ]
61
- ) -> Dict [str , Any ]:
64
+ async def execute_codemod (codemod : Annotated [str , "The python codemod code to execute on the codebase" ]) -> Dict [str , Any ]:
62
65
if not state .parse_task or not state .parse_task .done ():
63
66
return {"error" : "Codebase is not ready for codemod execution." }
64
-
67
+
65
68
try :
66
- codebase = await state .parse_task
69
+ await state .parse_task
67
70
# TODO: Implement proper sandboxing for code execution
68
71
context = {
69
72
"codebase" : state .parsed_codebase ,
@@ -72,23 +75,15 @@ async def execute_codemod(
72
75
exec (codemod , context )
73
76
74
77
logs = "\n " .join (state .log_buffer )
75
-
78
+
76
79
state .reset ()
77
- return {
78
- "message" : "Codemod executed and codebase reset." ,
79
- "logs" : logs
80
- }
80
+ return {"message" : "Codemod executed and codebase reset." , "logs" : logs }
81
81
except Exception as e :
82
- return {
83
- "error" : f"Error executing codemod: { str (e )} " ,
84
- "details" : {
85
- "type" : type (e ).__name__ ,
86
- "message" : str (e )
87
- }
88
- }
82
+ return {"error" : f"Error executing codemod: { str (e )} " , "details" : {"type" : type (e ).__name__ , "message" : str (e )}}
83
+
89
84
90
85
if __name__ == "__main__" :
91
- print (' starting codegen-mcp-server' )
86
+ print (" starting codegen-mcp-server" )
92
87
run = mcp .run_stdio_async ()
93
- print (' codegen-mcp-server started' )
88
+ print (" codegen-mcp-server started" )
94
89
asyncio .run (run )
0 commit comments