diff --git a/build.py b/build.py index 1d27f88..3b7ea70 100644 --- a/build.py +++ b/build.py @@ -81,7 +81,7 @@ def build(): '--onefile', '--version-file=version_info.txt', f'--add-data=pyproject.toml{sep}.', - f'--add-data=resources/app.png{sep}resources', + f'--add-data=resources{sep}resources', f'--name={name}', f'--icon={icon}', '--clean', diff --git a/main.py b/main.py index 6e68418..676bb86 100644 --- a/main.py +++ b/main.py @@ -65,6 +65,14 @@ def run(log_level: str = "CRITICAL"): instructions = """ # Perfecto MCP Server +Use Perfecto MCP (Model Context Protocol) to interact with Perfecto cloud services. +Use this MCP Server when you need to query devices, retrieve execution reports, access help documentation, +manage scriptless tests, explore available skills, or get user information programmatically through the MCP interface. + +Use perfecto_skills tool to discover Perfecto skills. +Start with perfecto-mcp-tools skill to understand available MCP tools. +Then discover and read relevant skills based on user queries. + """ mcp = FastMCP("perfecto-mcp", instructions=instructions, diff --git a/pyproject.toml b/pyproject.toml index a45dc3d..8df648d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,4 +26,4 @@ where = ["."] include = ["tools", "config", "models", "formatters", "resources"] [tool.setuptools.package-data] -"resources" = ["*.png"] +"resources" = ["**/*"] diff --git a/server.py b/server.py index c85535a..e8aafb1 100644 --- a/server.py +++ b/server.py @@ -5,6 +5,7 @@ from tools.device_manager import register as register_device_manager from tools.execution_manager import register as register_execution_manager from tools.help_manager import register as register_help_manager +from tools.skills_manager import register as register_skills_manager from tools.user_manager import register as register_user_manager @@ -20,4 +21,5 @@ def register_tools(mcp, token: Optional[PerfectoToken]): register_device_manager(mcp, token) register_execution_manager(mcp, token) register_help_manager(mcp, token) + register_skills_manager(mcp, token) register_ai_scriptless_manager(mcp, token) diff --git a/tools/skills_manager.py b/tools/skills_manager.py new file mode 100644 index 0000000..1fa7dec --- /dev/null +++ b/tools/skills_manager.py @@ -0,0 +1,150 @@ +import traceback +from typing import Optional, Dict, Any + +import httpx +from mcp.server.fastmcp import Context +from pydantic import Field + +from config.perfecto import TOOLS_PREFIX, SUPPORT_MESSAGE +from config.token import PerfectoToken +from models.manager import Manager +from models.result import BaseResult +from tools.skills_utils import list_skills, read_skill_definition, read_skill_file, parse_skill_uri, \ + is_skill_uri, list_skill_resources_uri + + +# This it's based on the ideas behind Anthropic Skills +# More info about Skills https://github.com/anthropics/skills + +class SkillsManager(Manager): + skills = None # Static to share between different instance of SkillsManager + + def __init__(self, token: Optional[PerfectoToken], ctx: Context): + super().__init__(token, ctx) + + @staticmethod + async def list_skills() -> BaseResult: + errors = [] + if SkillsManager.skills is None: + skills, errors = list_skills() + SkillsManager.skills = skills + + return BaseResult( + result=SkillsManager.skills, + error=errors[0] if errors and len(errors) > 0 else None # Only the first error + ) + + @staticmethod + async def read_skill(skill_name: str) -> BaseResult: + skill_content, error = read_skill_definition(skill_name) + return BaseResult( + result={ + "skill_name": skill_name, + "path": "SKILL.md", + "content": skill_content, + }, + error=error + ) + + @staticmethod + async def read_skill_file_path(skill_name: str, file_path: str) -> BaseResult: + skill_content, error = read_skill_file(skill_name, file_path) + return BaseResult( + result={ + "skill_name": skill_name, + "path": file_path, + "content": skill_content, + }, + error=error + ) + + @staticmethod + async def list_skill_resources(skill_name: str) -> BaseResult: + skill_resources = list_skill_resources_uri(skill_name) + return BaseResult( + result={ + "skill_name": skill_name, + "resources": skill_resources, + } + ) + + @staticmethod + async def read_skill_resource_uri(skill_uri: str) -> BaseResult: + if is_skill_uri(skill_uri): + skill_name, file_path = parse_skill_uri(skill_uri) + skill_content, error = read_skill_file(skill_name, file_path) + return BaseResult( + result={ + "skill_name": skill_name, + "path": file_path, + "content": skill_content, + }, + error=error + ) + else: + return BaseResult( + error=f"Invalid Skill URI: {skill_uri}" + ) + + +def register(mcp, token: Optional[PerfectoToken]): + @mcp.resource("skills-{skill_name}://{path}") + def universal_skills_handler(skill_name: str, path: str) -> BaseResult: + content, error = read_skill_file(skill_name, path) + return BaseResult( + result={ + "skill_name": skill_name, + "path": path, + "content": content, + }, + error=error + ) + + @mcp.tool( + name=f"{TOOLS_PREFIX}_skills", + description=""" +Operations to obtain Skills around Perfecto. +Actions: +- list_skills: List all the Skills available to learn. +- read_skill: Read detailed information about a specific skill_name. + args(dict): Dictionary with the following required parameters: + skill_name (str): The skill name. +- list_skill_resources: List all the Skills Resources available to learn. + args(dict): Dictionary with the following required parameters: + skill_name (str): The skill name. +- read_skill_resource_uri: Read file content based on a Skill Resource URI (skill-{skill_name}://{resource_path}). + args(dict): Dictionary with the following required parameters: + skill_resource_uri (str): The skill URI. + +""" + ) + async def skills( + action: str = Field(description="The action id to execute"), + args: Dict[str, Any] = Field(description="Dictionary with parameters", default=None), + ctx: Context = Field(description="Context object providing access to MCP capabilities") + ) -> BaseResult: + if args is None: + args = {} + skills_manager = SkillsManager(token, ctx) + try: + match action: + case "list_skills": + return await skills_manager.list_skills() + case "read_skill": + return await skills_manager.read_skill(args.get("skill_name", "")) + case "list_skill_resources": + return await skills_manager.list_skill_resources(args.get("skill_name", "")) + case "read_skill_resource_uri": + return await skills_manager.read_skill_resource_uri(args.get("skill_resource_uri", "")) + case _: + return BaseResult( + error=f"Action {action} not found in skills manager tool" + ) + except httpx.HTTPStatusError: + return BaseResult( + error=f"Error: {traceback.format_exc()}" + ) + except Exception: + return BaseResult( + error=f"Error: {traceback.format_exc()}\n{SUPPORT_MESSAGE}" + ) diff --git a/tools/skills_utils.py b/tools/skills_utils.py new file mode 100644 index 0000000..2e24700 --- /dev/null +++ b/tools/skills_utils.py @@ -0,0 +1,225 @@ +import os +import re +from pathlib import Path +from typing import Tuple, Dict, List + +from tools.utils import get_resources_path + +SKILL_URI_REGEX = re.compile(r"^skill-(?P[a-zA-Z0-9_-]+)://(?P.+)$") + + +def parse_frontmatter(frontmatter: str) -> Dict[str, str]: + meta = {} + lines = frontmatter.splitlines() + current_key = None + current_value = [] + block_mode = False + + for line in lines: + stripped = line.strip() + if not stripped or stripped.startswith('#'): + continue # Skip empty or comment lines + + if ':' in line and not line.startswith((' ', '\t')): # New key at root level + if current_key: + value_str = '\n'.join(current_value).strip() + value_str = str(value_str).strip('\'"') # Strip outer quotes if present + meta[current_key] = value_str + try: + key, value = [part.strip() for part in line.split(':', 1)] + value = str(value).strip('\'"') # Strip quotes from single-line value + except ValueError: + raise ValueError("Invalid key-value format in frontmatter") + + if key not in ('name', 'description'): + continue # Ignore unexpected keys + + if value in ('|', '>'): + block_mode = True + current_value = [] + else: + block_mode = False + current_value = [value] + + current_key = key + elif current_key and (block_mode or line.startswith((' ', '\t'))): + # Append to multi-line block (indented or in block mode) + current_value.append(line.rstrip() if block_mode else line.strip()) + else: + raise ValueError("Malformed frontmatter structure") + + if current_key: + value_str = '\n'.join(current_value).strip() + value_str = value_str.strip('\'"') # Strip outer quotes if present + meta[current_key] = value_str + + # Ensure required keys are present after parsing + if 'name' not in meta or 'description' not in meta: + raise ValueError("Missing 'name' or 'description' in frontmatter") + + return meta + + +def list_skills() -> Tuple[List[Dict], List[str]]: + full_skills_dir = os.path.join(get_resources_path(), 'skills') + skills = [] + errors = [] + skills_path = Path(full_skills_dir) + + if not os.path.exists(skills_path): + # Graceful handling + return [], [f"Missing skills directory: {skills_path}"] + + for folder in os.listdir(skills_path): + folder_path = skills_path / folder + if folder_path.is_dir(): + md_path = folder_path / 'SKILL.md' + if md_path.exists(): + skill, error = read_skill_meta(md_path) + if skill is not None: + skills.append(skill) + if error is not None: + errors.append(error) + else: + errors.append(f"Skill file not found: {md_path}") + else: + errors.append(f"Invalid Skill folder {folder_path}") + return skills, errors + + +def validate_skill_content(content, md_path): + if not content.startswith('---'): + return False, "No YAML frontmatter found" + + match = re.match(r'^---\n(.*?)\n---', content, re.DOTALL) + if not match: + return False, "Invalid frontmatter format" + + frontmatter = match.group(1) + + if 'name:' not in frontmatter: + return False, "Missing 'name' in frontmatter" + if 'description:' not in frontmatter: + return False, "Missing 'description' in frontmatter" + + try: + meta = parse_frontmatter(frontmatter) + name = meta.get('name') + description = meta.get('description') + + if not name or not description: + return False, "Name or description is empty" + + # Validate name: hyphen-case + if not re.match(r'^[a-z0-9-]+$', name): + return False, f"Name '{name}' should be hyphen-case (lowercase letters, digits, and hyphens only)" + if name.startswith('-') or name.endswith('-') or '--' in name: + return False, f"Name '{name}' cannot start/end with hyphen or contain consecutive hyphens" + + # Validate description: no angle brackets + if '<' in description or '>' in description: + return False, "Description cannot contain angle brackets (< or >)" + + except ValueError as e: + return False, f"Error parsing frontmatter in {md_path}: {str(e)}" + + return True, "Skill is valid" + + +def read_skill_meta(md_path) -> Tuple[Dict[str, None], str | None]: + skill = None + error = None + content = md_path.read_text(encoding='utf-8') + valid, message = validate_skill_content(content, md_path) + if valid: + match = re.match(r'^---\n(.*?)\n---', content, re.DOTALL) + if match: + frontmatter = match.group(1) + try: + meta = parse_frontmatter(frontmatter) + name = meta.get('name') + description = meta.get('description') + if name and description: + skill = { + 'name': name, + 'description': description + } + except ValueError as e: + error = f"Error parsing frontmatter in {md_path}: {str(e)}" + else: + error = f"Validation failed for {md_path}: {message}" + return skill, error + + +def get_skill_file_path(skill_name: str, file_path: str) -> Path: + full_skills_dir = os.path.join(get_resources_path(), 'skills') + skills_path = Path(full_skills_dir) + return skills_path / skill_name / file_path + + +def replace_skills_markdown_links(content: str, skill_name: str, file_path: str) -> str: + # Markdown Link Pattern []() + pattern = re.compile(r'\[([^\]]+)\]\(([^)]+)\)') + + def replacer(match): + text, url = match.groups() + # Detect when not it's a protocol link + if not re.match(r'^[a-zA-Z][a-zA-Z0-9+.-]*://', url): + # Only when it's relative to local, generate the uri format + # Transform the relative path to absolute and cut the base path + skills_path = Path(os.path.join(get_resources_path(), 'skills')) + base_path = skills_path / skill_name / file_path + url = (base_path.parent / url).resolve() + try: + url = url.relative_to(skills_path / skill_name).as_posix() + new_url = f"skill-{skill_name}://{url}" + return f"[{text}]({new_url})" + except: + return match.group(0) # In case the relative not it's valid, use the original + else: + return match.group(0) # Other case return the original + + return pattern.sub(replacer, content) + + +def read_skill_file(skill_name: str, file_path: str) -> Tuple[str | None, str | None]: + skill_file_path = get_skill_file_path(skill_name, file_path) + if skill_file_path.exists(): + content = skill_file_path.read_text(encoding='utf-8') + if file_path.endswith('.md'): # Only on Markdown replace to skills uri format + content = replace_skills_markdown_links(content, skill_name, file_path) + if file_path.endswith('SKILL.md'): + valid, message = validate_skill_content(content, file_path) + if valid: + return content, None + else: + return None, f"Validation failed for {skill_name}: {message}" + else: + return content, None + else: + return None, f"Skill file not found: {skill_file_path}" + + +def is_skill_uri(uri: str) -> bool: + return bool(SKILL_URI_REGEX.match(uri)) + + +def parse_skill_uri(uri: str) -> Tuple[str, str]: + match = SKILL_URI_REGEX.match(uri) + if not match: + raise ValueError(f"Invalid Skill URI : {uri}") + return match.group("skill_name"), match.group("path") + +def list_skill_resources_uri(skill_name: str) -> List[str]: + full_skills_dir = os.path.join(get_resources_path(), 'skills') + skills_path = Path(full_skills_dir) + skill_path = skills_path / skill_name + skill_resources = [] + for file_path in skill_path.rglob("*"): + if file_path.is_file(): + url = file_path.relative_to(skills_path / skill_name).as_posix() + skill_resources.append(f"skill-{skill_name}://{url}") + return skill_resources + +def read_skill_definition(skill_name: str) -> Tuple[str | None, str | None]: + return read_skill_file(skill_name, 'SKILL.md')