|
| 1 | +import json |
| 2 | +from typing import Annotated, Any |
| 3 | +from mcp.server.fastmcp import FastMCP, Context |
| 4 | +from codegen import Codebase |
| 5 | +from codegen.sdk.codebase.span import Span |
| 6 | +from codegen.cli.api.client import RestAPI |
| 7 | +import codegen |
| 8 | +# Initialize FastMCP server |
| 9 | +mcp = FastMCP("codegen-mcp") |
| 10 | + |
| 11 | + |
| 12 | + |
| 13 | +def get_codebase_path(dir_path: str) -> Codebase: |
| 14 | + """Get the path to the codebase.""" |
| 15 | + codebase = Codebase(dir_path) |
| 16 | + return codebase |
| 17 | + |
| 18 | + |
| 19 | +@mcp.resource("config://service", mime_type="application/json") |
| 20 | +def get_service_config() -> dict[str, Any]: |
| 21 | + """Get the service config.""" |
| 22 | + return { |
| 23 | + "name": "codegen-mcp", |
| 24 | + "version": "0.1.0", |
| 25 | + "description": "A service for generating codemods for a given task and codebase using the codegen sdk", |
| 26 | + } |
| 27 | + |
| 28 | +# @mcp.tool() |
| 29 | +# def get_all_functions_in_filepath(codebase_path: Annotated[str, "The absolute path to the codebase directory"], file_path: Annotated[str, "The codebase relative path to the file to get functions from"]) -> list[str]: |
| 30 | +# """Get all functions in the codebase.""" |
| 31 | +# codebase = get_codebase_path(codebase_path) |
| 32 | +# file = codebase.get_file(file_path) |
| 33 | + |
| 34 | +# return [func.span for func in file.function_calls] |
| 35 | + |
| 36 | + |
| 37 | +# @mcp.tool() |
| 38 | +# def get_all_usages_for_symbol(codebase_path: Annotated[str, "The absolute path to the codebase directory"], symbol_name: Annotated[str, "The name of the symbol to get usages for"]) -> list[str]: |
| 39 | +# """Get all files in the codebase, given the absolute path to codebase directory and the name of the symbol""" |
| 40 | +# codebase = get_codebase_path(codebase_path) |
| 41 | +# target = codebase.get_symbol(symbol_name) |
| 42 | + |
| 43 | +# if hasattr(target, 'usages'): |
| 44 | +# return [usage.usage_symbol.span for usage in target.usages] |
| 45 | +# else: |
| 46 | +# return [] |
| 47 | + |
| 48 | + |
| 49 | +@mcp.tool() |
| 50 | +def generate_codemod(task: Annotated[str, "The task to which the codemod should implement to solve"], codebase_path: Annotated[str, "The absolute path to the codebase directory"], ctx: Context) -> str: |
| 51 | + """Generate a codemod for the given task and codebase.""" |
| 52 | + codebase = get_codebase_path(codebase_path) |
| 53 | + language = codebase.language |
| 54 | + return f'''use the codegen cli and run the following command: |
| 55 | + |
| 56 | + codegen create generate-codemod -d "{task}" |
| 57 | + |
| 58 | + ''' |
| 59 | + |
| 60 | + |
| 61 | +@mcp.tool() |
| 62 | +def get_codemod_for_prompt(prompt: Annotated[str, "The task to which the codemod should implement to solve"], include_files: Annotated[list[str], "The files to include in the codemod"], exclude_files: Annotated[list[str], "The files to exclude from the codemod"], ctx: Context) -> str: |
| 63 | + """Get the path to the codebase.""" |
| 64 | + codebase = Codebase(codebase_path) |
| 65 | + return codebase.path |
| 66 | + |
| 67 | + |
| 68 | +@mcp.tool() |
| 69 | +def improve_codemod( |
| 70 | + codemod_source: Annotated[str, "The source code of the codemod to improve"], |
| 71 | + concerns: Annotated[list[str], "A list of issues that were discovered with the current codemod that need to be considered in the next iteration"], |
| 72 | + context: Annotated[dict[str, Any], "Additional context for the codemod this can be a list of files that are related, additional information about the task, etc."], |
| 73 | + ctx: Context |
| 74 | +) -> str: |
| 75 | + """Improve the codemod.""" |
| 76 | + return f'''use the codegen cli and run the following command: |
| 77 | + |
| 78 | + codegen create improve-codemod -d "{codemod_source}" |
| 79 | + |
| 80 | + ''' |
| 81 | + |
| 82 | +if __name__ == "__main__": |
| 83 | + # Initialize and run the server |
| 84 | + |
| 85 | + print("Starting codegen server...") |
| 86 | + mcp.run(transport='stdio') |
0 commit comments