Skip to content
Merged

Editing #40814

Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 12 additions & 1 deletion .vscode/mcp.json
Original file line number Diff line number Diff line change
@@ -1,4 +1,12 @@
{
"inputs": [
{
"type": "promptString",
"id": "gh_token",
"description": "GitHub Token",
"password": true
}
],
"servers": {
"typespec-python": {
"command": "uv",
Expand All @@ -7,7 +15,10 @@
"${workspaceFolder}/tools/mcp/typespec-story/",
"run",
"main.py"
]
],
"env": {
"GH_TOKEN": "${input:gh_token}"
},
},
"azure-sdk-validation": {
"command": "uv",
Expand Down
1 change: 0 additions & 1 deletion tools/mcp/typespec-story/.python-version

This file was deleted.

269 changes: 79 additions & 190 deletions tools/mcp/typespec-story/main.py
Original file line number Diff line number Diff line change
@@ -1,16 +1,77 @@
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)
hander = logging.StreamHandler(sys.stderr)
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://(?P<urlRoot>github|raw.githubusercontent).com/(?P<repo>[^/]*/azure-rest-api-specs(-pr)?)/(tree/|blob/)?(?P<commit>[0-9a-f]{40}|[^/]+)/(?P<path>.*)/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]:
Expand All @@ -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)}")

Expand All @@ -40,203 +92,40 @@ 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,
"stdout": result.stdout,
"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__":
Expand Down
3 changes: 2 additions & 1 deletion tools/mcp/typespec-story/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -2,14 +2,15 @@
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 = [
{name = "Azure SDK Team", email = "azpysdk@microsoft.com"},
]
dependencies = [
"mcp[cli]>=1.6.0",
"pygithub>=2.6.1",
]