diff --git a/.vscode/mcp.json b/.vscode/mcp.json index 1fbe01e66f20..b8c31695a45e 100644 --- a/.vscode/mcp.json +++ b/.vscode/mcp.json @@ -1,4 +1,12 @@ { + "inputs": [ + { + "type": "promptString", + "id": "gh_token", + "description": "GitHub Token", + "password": true + } + ], "servers": { "typespec-python": { "command": "uv", @@ -7,7 +15,10 @@ "${workspaceFolder}/tools/mcp/typespec-story/", "run", "main.py" - ] + ], + "env": { + "GH_TOKEN": "${input:gh_token}" + }, }, "azure-sdk-validation": { "command": "uv", diff --git a/tools/mcp/typespec-story/.python-version b/tools/mcp/typespec-story/.python-version deleted file mode 100644 index e4fba2183587..000000000000 --- a/tools/mcp/typespec-story/.python-version +++ /dev/null @@ -1 +0,0 @@ -3.12 diff --git a/tools/mcp/typespec-story/main.py b/tools/mcp/typespec-story/main.py index 918641b23a59..d96d89edded7 100644 --- a/tools/mcp/typespec-story/main.py +++ b/tools/mcp/typespec-story/main.py @@ -1,8 +1,11 @@ +import re from mcp.server.fastmcp import FastMCP import subprocess -from typing import Dict, Optional, Any +from typing import Dict, Optional, Any, Match, Union import logging import sys +import os +from github import Github, Auth logger = logging.getLogger() logger.setLevel(logging.INFO) @@ -10,7 +13,65 @@ logger.addHandler(hander) # Initialize FastMCP server -mcp = FastMCP("typespec-story") +mcp = FastMCP("typespec") + +def get_latest_commit(tspurl: str) -> Union[Match[str], None]: + """Get the latest commit hash for a given TypeSpec config URL. + + Args: + tspurl: The URL to the tspconfig.yaml file. + + Returns: + The latest commit hash for the specified TypeSpec configuration. + """ + # Extract the service name, repo, commit, and tspconfig path from the URL + try: + # Use regex to validate and extract components from the URL + configUrl = tspurl + res = re.match( + r"^https://(?Pgithub|raw.githubusercontent).com/(?P[^/]*/azure-rest-api-specs(-pr)?)/(tree/|blob/)?(?P[0-9a-f]{40}|[^/]+)/(?P.*)/tspconfig.yaml$", + configUrl + ) + + commit = None + if res: + groups = res.groupdict() + commit = groups["commit"] + logger.info(f"Extracted commit: {commit}") + + # Parse the URL to extract the path within the repository + repo_parts = tspurl.split("/") + repo_name = f"{repo_parts[3]}/{repo_parts[4]}" + + parts = tspurl.split("azure-rest-api-specs/blob/")[1].split("/") + parts.pop(0) # Remove the branch name (e.g., 'main') + + # Join all parts until the last directory (containing tspconfig.yaml) + folder_path = "/".join(parts) + logger.info(f"Extracted folder path from URL: {folder_path}") + + auth = Auth.Token(os.getenv("GH_TOKEN")) + g = Github(auth=auth) + logger.info(f"Authenticated to GitHub with token: {g}") + repo = g.get_repo(repo_name) + + # Get commits that affect the specific folder + # TODO: if commit is a branch name + if not commit or commit == "main": + commits = repo.get_commits(path=folder_path) + + if not commits: + raise ValueError(f"No commits found for path: {folder_path}") + + latest_commit = commits[0].sha + logger.info(f"Found latest commit for {latest_commit}") + return f"https://raw.githubusercontent.com/{groups['repo']}/{latest_commit}/{groups['path']}/tspconfig.yaml" + return f"https://raw.githubusercontent.com/{groups['repo']}/{commit}/{groups['path']}/tspconfig.yaml" + + except Exception as e: + logger.error(f"An error occurred: {str(e)}") + raise + # Helper function to run CLI commands def run_typespec_cli_command(command: str, args: Dict[str, Any], root_dir: Optional[str] = None) -> Dict[str, Any]: @@ -19,17 +80,8 @@ def run_typespec_cli_command(command: str, args: Dict[str, Any], root_dir: Optio # Convert args dict to CLI arguments for key, value in args.items(): - if key == "_": - # Handle positional arguments - if isinstance(value, list): - cli_args.extend(value) - else: - cli_args.append(value) - elif value is True: - cli_args.append(f"--{key}") - elif value is not False and value is not None: - cli_args.append(f"--{key}") - cli_args.append(str(value)) + cli_args.append(f"--{key}") + cli_args.append(str(value)) logger.info(f"Running command: {' '.join(cli_args)}") @@ -40,8 +92,8 @@ def run_typespec_cli_command(command: str, args: Dict[str, Any], root_dir: Optio check=True, capture_output=True, text=True, - cwd=root_dir, ) + logger.info(f"Command output: {result.stdout}") return { "success": True, @@ -49,194 +101,31 @@ def run_typespec_cli_command(command: str, args: Dict[str, Any], root_dir: Optio "stderr": result.stderr, "code": result.returncode } - except subprocess.CalledProcessError as e: - return { - "success": False, - "stdout": e.stdout, - "stderr": e.stderr, - "code": e.returncode, - "error": str(e) - } + except Exception as e: + logger.error(f"Command failed with error: {str(e)}") + raise + # Register tools for each TypeSpec client generator CLI command @mcp.tool("init") -def init_tool(tspconfig_url: Optional[str] = None, root_dir: Optional[str] = None) -> Dict[str, Any]: - """Initialize a client library directory using a tspconfig.yaml file. +def init_tool(tsp_config_url: str, root_dir: Optional[str] = None) -> Dict[str, Any]: + """Initialize a typespec client library directory given the url. Args: - tspconfig_url: The URL of the tspconfig.yaml file. - root_dir: The root directory where the client library will be generated. # TODO: Unsure if this is needed + tsp_config_url: The URL to the tspconfig.yaml file. + root_dir: The root directory where the client library will be generated. Returns: A dictionary containing the result of the command. """ - args = {} - if tspconfig_url: - args["tsp-config"] = tspconfig_url - - return run_typespec_cli_command("init", args, root_dir=root_dir) - -@mcp.tool("update") -def update_tool(path: Optional[str] = None) -> Dict[str, Any]: - """Look for a tsp-location.yaml file to sync a TypeSpec project and generate a client library. - - Args: - path: Path to a tsp-location.yaml file or a directory containing one. + # Get the URL to the tspconfig.yaml file + updated_url = get_latest_commit(tsp_config_url) - Returns: - A dictionary containing the result of the command. - """ + # Prepare arguments for the CLI command args = {} - if path: - args["_"] = [path] + args["tsp-config"] = updated_url - return run_typespec_cli_command("update", args) - -@mcp.tool("sync") -def sync_tool(repo_url: str, commit: Optional[str] = None, tag: Optional[str] = None, - branch: Optional[str] = None, path: Optional[str] = None) -> Dict[str, Any]: - """Sync a TypeSpec project with parameters specified in tsp-location.yaml. - - Args: - repo_url: URL of the repository containing the TypeSpec project. - commit: Commit hash to sync to. - tag: Tag to sync to. - branch: Branch to sync to. - path: Path within the repository to the TypeSpec project. - - Returns: - A dictionary containing the result of the command. - """ - args = { - "_": [repo_url] - } - - if commit: - args["commit"] = commit - if tag: - args["tag"] = tag - if branch: - args["branch"] = branch - if path: - args["path"] = path - - return run_typespec_cli_command("sync", args) - -@mcp.tool("generate") -def generate_tool(project_directory: str, output_directory: Optional[str] = None, - language: Optional[str] = None, tsp_emit: Optional[str] = None, - service_dir: Optional[str] = None) -> Dict[str, Any]: - """Generate a client library from a TypeSpec project. - - Args: - project_directory: The directory containing the TypeSpec project. - output_directory: The directory where the client library will be generated. - language: The language to generate client code for. - tsp_emit: The emitter to use for generating the client library. - service_dir: The directory containing the service definition. - - Returns: - A dictionary containing the result of the command. - """ - args = { - "_": [project_directory] - } - - if output_directory: - args["output-dir"] = output_directory - if language: - args["language"] = language - if tsp_emit: - args["tsp-emit"] = tsp_emit - if service_dir: - args["service-dir"] = service_dir - - return run_typespec_cli_command("generate", args) - -@mcp.tool("convert") -def convert_tool(swagger_path: str, output_directory: Optional[str] = None, - title: Optional[str] = None, version: Optional[str] = None) -> Dict[str, Any]: - """Convert an existing swagger specification to a TypeSpec project. - - Args: - swagger_path: Path to the Swagger specification file. - output_directory: The directory where the TypeSpec project will be generated. - title: The title for the generated TypeSpec project. - version: The version for the generated TypeSpec project. - - Returns: - A dictionary containing the result of the command. - """ - args = { - "_": [swagger_path] - } - - if output_directory: - args["output-dir"] = output_directory - if title: - args["title"] = title - if version: - args["version"] = version - - return run_typespec_cli_command("convert", args) - -@mcp.tool("compare") -def compare_tool(old_swagger: str, new_swagger: str, output_file: Optional[str] = None) -> Dict[str, Any]: - """Compare two Swagger definitions to identify relevant differences. - - Args: - old_swagger: Path to the old Swagger specification file. - new_swagger: Path to the new Swagger specification file. - output_file: Path to the output file where the comparison results will be written. - - Returns: - A dictionary containing the result of the command. - """ - args = { - "_": [old_swagger, new_swagger] - } - - if output_file: - args["output-file"] = output_file - - return run_typespec_cli_command("compare", args) - -@mcp.tool("sort_swagger") -def sort_swagger_tool(swagger_path: str, output_file: Optional[str] = None) -> Dict[str, Any]: - """Sort an existing swagger specification to match TypeSpec generated swagger order. - - Args: - swagger_path: Path to the Swagger specification file. - output_file: Path to the output file where the sorted Swagger will be written. - - Returns: - A dictionary containing the result of the command. - """ - args = { - "_": [swagger_path] - } - - if output_file: - args["output-file"] = output_file - - return run_typespec_cli_command("sort-swagger", args) - -@mcp.tool("generate_config_files") -def generate_config_files_tool(output_directory: Optional[str] = None) -> Dict[str, Any]: - """Generate default configuration files used by tsp-client. - - Args: - output_directory: The directory where the configuration files will be generated. - - Returns: - A dictionary containing the result of the command. - """ - args = {} - - if output_directory: - args["output-dir"] = output_directory - - return run_typespec_cli_command("generate-config-files", args) + return run_typespec_cli_command("init", args, root_dir=root_dir) # Run the MCP server if __name__ == "__main__": diff --git a/tools/mcp/typespec-story/pyproject.toml b/tools/mcp/typespec-story/pyproject.toml index eb52765a7adf..fb6317153951 100644 --- a/tools/mcp/typespec-story/pyproject.toml +++ b/tools/mcp/typespec-story/pyproject.toml @@ -2,7 +2,7 @@ name = "typespec-story-mcp" version = "0.1.0" description = "MCP server for TypeSpec client generator CLI" -requires-python = ">=3.12" +requires-python = ">=3.10" license = "MIT" readme = "README.md" authors = [ @@ -10,6 +10,7 @@ authors = [ ] dependencies = [ "mcp[cli]>=1.6.0", + "pygithub>=2.6.1", ]