| " + header + " | " + header_html = "
|---|
| " + cell.replace("\n", " ") + " | "
+ markdown.append(cells_html)
+ else:
+ markdown.append("| " + " | ".join(cell_texts) + " |")
+ if as_html:
+ markdown.append("
{text}")
+ else:
+ parts.append(f"`{text}`")
+ else:
+ inner_result = process_inline_elements(child, base_url)
+ if inner_result:
+ parts.append(inner_result)
+
+ if child.tail:
+ parts.append(child.tail)
+
+ return ''.join(parts)
+
+
+def element_to_markdown(element, base_url=None, level=0):
+ code_block_lang = ['javascript', 'java', 'python', 'ruby', 'go', 'php', 'c#', 'csharp', 'typescript',
+ 'bash', 'shell', 'sql', 'json', 'xml', 'yaml', 'css', 'html']
+
+ tag = element.tag.lower()
+ result = []
+
+ # Headers
+ if tag in ['h1', 'h2', 'h3', 'h4', 'h5', 'h6']:
+ level_num = int(tag[1])
+ text = clean_text(element.text_content())
+ if text:
+ result.append(f"{'#' * level_num} {text}\n")
+
+ elif tag == 'p':
+ text = clean_text(process_inline_elements(element, base_url))
+ if text:
+ result.append(f"{text}\n")
+
+ elif tag == 'ul':
+ for li in element.xpath('./li'):
+ text = clean_text(process_inline_elements(li, base_url))
+ if text:
+ result.append(f"- {text}\n")
+ result.append("")
+
+ elif tag == 'ol':
+ for i, li in enumerate(element.xpath('./li'), 1):
+ text = clean_text(process_inline_elements(li, base_url))
+ if text:
+ result.append(f"{i}. {text}\n")
+ result.append("")
+
+ elif tag == 'table':
+ table_md = table_to_markdown(element, base_url)
+ if table_md:
+ result.append(f"{table_md}\n")
+
+ elif tag == 'pre' or 'codesnippet' in element.get('class', '').lower():
+ lang = ""
+ code_element = element
+
+ # Try to get the language code and the source code
+
+ lang_elem = element.xpath('.//*[contains(@class, "language")]')
+ if not lang_elem:
+ for child in element:
+ child_text = child.text_content().strip().lower()
+ if child_text in code_block_lang:
+ lang = child_text
+ break
+
+ code_children = element.xpath('.//code')
+ if code_children:
+ code_element = code_children[0]
+ class_attr = code_element.get('class', '')
+ if 'language-' in class_attr:
+ lang = class_attr.split('language-')[1].split()[0]
+
+ code_text = extract_text_with_br(code_element)
+ code_text = clean_text(code_text, preserve_newlines=True)
+
+ lines = code_text.split('\n')
+ filtered_lines = []
+
+ for i, line in enumerate(lines):
+ line_stripped = line.strip().lower()
+ if line_stripped == 'copy': # Exclude Copy element (UI Element)
+ continue
+ if line_stripped in code_block_lang:
+ lang = line_stripped
+
+ filtered_lines.append(line)
+
+ code_text = '\n'.join(filtered_lines).strip()
+
+ if code_text:
+ result.append(f"```{lang}\n{code_text}\n```\n")
+
+ elif tag == 'blockquote':
+ text = extract_text_with_br(element)
+ text = clean_text(text)
+ if text:
+ for line in text.split('\n'):
+ if line.strip():
+ result.append(f"> {line}\n")
+ result.append("")
+
+ elif tag == 'hr':
+ result.append("---\n")
+
+ elif tag == 'img':
+ alt = element.get('alt', '')
+ src = element.get('src', '')
+ if src and base_url:
+ src = urljoin(base_url, src)
+ result.append(f"\n")
+
+ elif tag in ['script', 'style', 'noscript', 'meta', 'link', 'head']:
+ # Ignore
+ pass
+
+ # For any others elements, process the children
+ else:
+ for child in element:
+ child_md = element_to_markdown(child, base_url, level + 1)
+ if child_md:
+ result.extend(child_md)
+
+ return result
+
+
+def html_to_markdown(html_content, base_url=None):
+ tree = lxml.html.fromstring(html_content)
+
+ main_div = tree.xpath('//div[@role="main"]')
+ if not main_div:
+ main_div = tree.xpath('//main | //div[@class="main"] | //article | //body')
+
+ if not main_div:
+ return "# Error\n\nMain content not found"
+
+ main_div = main_div[0]
+
+ markdown_lines = []
+ for child in main_div:
+ result = element_to_markdown(child, base_url)
+ if result:
+ markdown_lines.extend(result)
+
+ markdown = "".join(markdown_lines)
+
+ while "\n\n\n" in markdown:
+ markdown = markdown.replace("\n\n\n", "\n\n")
+
+ return markdown.strip()
+
+
+def convert_js_to_py_dict(js_text: str) -> dict:
+ # Convert javascript dictionary to python dictionary
+ js_text = js_text.replace("define(", "").replace(");", "").replace("'", '"')
+ js_text = re.sub(r'//.*', '', js_text)
+ js_text = re.sub(r'/\*[\s\S]*?\*/', '', js_text)
+ js_text = re.sub(r',\s*(?=[}\]])', '', js_text)
+ js_text = re.sub(r'([{\[,]\s*)([A-Za-z_][A-Za-z0-9_]*)\s*:', r'\1"\2":', js_text)
+ return json.loads(js_text)
diff --git a/tools/user_manager.py b/tools/user_manager.py
index 7465dca..f6f816d 100644
--- a/tools/user_manager.py
+++ b/tools/user_manager.py
@@ -36,10 +36,11 @@ def register(mcp, token: Optional[PerfectoToken]):
)
async def user(
action: str = Field(description="The action id to execute"),
- args: Dict[str, Any] = Field(description="Dictionary with parameters"),
+ 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 = {}
user_manager = UserManager(token, ctx)
try:
match action:
diff --git a/tools/utils.py b/tools/utils.py
index 8920fb4..f29a3ae 100644
--- a/tools/utils.py
+++ b/tools/utils.py
@@ -1,15 +1,16 @@
"""
Simple utilities for Perfecto MCP tools.
"""
-import json
+import base64
+import os
import platform
-import re
+import sys
from datetime import datetime
+from importlib import resources
+from pathlib import Path
from typing import Optional, Callable
-from urllib.parse import urljoin
import httpx
-import lxml.html
from config.token import PerfectoToken
from config.version import __version__
@@ -106,300 +107,20 @@ def get_date_time_iso(timestamp: int) -> Optional[str]:
else:
return datetime.fromtimestamp(timestamp).isoformat()
-
-def clean_text(text, preserve_newlines=False):
- text = text.replace('\xa0', ' ')
-
- if preserve_newlines:
- lines = text.split('\n')
- cleaned_lines = [' '.join(line.split()) for line in lines]
- return '\n'.join(cleaned_lines).strip()
- else:
- text = ' '.join(text.split())
- return text.strip()
-
-
-def extract_text_with_br(element):
- html_str = lxml.html.tostring(element, encoding='unicode', method='html')
- html_str = html_str.replace('| " + header + " | " - header_html = "
|---|
| " + cell.replace("\n", " ") + " | "
- markdown.append(cells_html)
- else:
- markdown.append("| " + " | ".join(cell_texts) + " |")
- if as_html:
- markdown.append("
{text}")
- else:
- parts.append(f"`{text}`")
+def get_resources_path():
+ try:
+ resources_path = resources.files("resources")
+ except ModuleNotFoundError:
+ # Fallback for development or if not installed as package
+ if getattr(sys, 'frozen', False):
+ base_path = sys._MEIPASS
else:
- inner_result = process_inline_elements(child, base_url)
- if inner_result:
- parts.append(inner_result)
-
- if child.tail:
- parts.append(child.tail)
-
- return ''.join(parts)
-
-
-def element_to_markdown(element, base_url=None, level=0):
- code_block_lang = ['javascript', 'java', 'python', 'ruby', 'go', 'php', 'c#', 'csharp', 'typescript',
- 'bash', 'shell', 'sql', 'json', 'xml', 'yaml', 'css', 'html']
-
- tag = element.tag.lower()
- result = []
-
- # Headers
- if tag in ['h1', 'h2', 'h3', 'h4', 'h5', 'h6']:
- level_num = int(tag[1])
- text = clean_text(element.text_content())
- if text:
- result.append(f"{'#' * level_num} {text}\n")
-
- elif tag == 'p':
- text = clean_text(process_inline_elements(element, base_url))
- if text:
- result.append(f"{text}\n")
-
- elif tag == 'ul':
- for li in element.xpath('./li'):
- text = clean_text(process_inline_elements(li, base_url))
- if text:
- result.append(f"- {text}\n")
- result.append("")
-
- elif tag == 'ol':
- for i, li in enumerate(element.xpath('./li'), 1):
- text = clean_text(process_inline_elements(li, base_url))
- if text:
- result.append(f"{i}. {text}\n")
- result.append("")
-
- elif tag == 'table':
- table_md = table_to_markdown(element, base_url)
- if table_md:
- result.append(f"{table_md}\n")
-
- elif tag == 'pre' or 'codesnippet' in element.get('class', '').lower():
- lang = ""
- code_element = element
-
- # Try to get the language code and the source code
-
- lang_elem = element.xpath('.//*[contains(@class, "language")]')
- if not lang_elem:
- for child in element:
- child_text = child.text_content().strip().lower()
- if child_text in code_block_lang:
- lang = child_text
- break
-
- code_children = element.xpath('.//code')
- if code_children:
- code_element = code_children[0]
- class_attr = code_element.get('class', '')
- if 'language-' in class_attr:
- lang = class_attr.split('language-')[1].split()[0]
-
- code_text = extract_text_with_br(code_element)
- code_text = clean_text(code_text, preserve_newlines=True)
-
- lines = code_text.split('\n')
- filtered_lines = []
-
- for i, line in enumerate(lines):
- line_stripped = line.strip().lower()
- if line_stripped == 'copy': # Exclude Copy element (UI Element)
- continue
- if line_stripped in code_block_lang:
- lang = line_stripped
-
- filtered_lines.append(line)
-
- code_text = '\n'.join(filtered_lines).strip()
-
- if code_text:
- result.append(f"```{lang}\n{code_text}\n```\n")
-
- elif tag == 'blockquote':
- text = extract_text_with_br(element)
- text = clean_text(text)
- if text:
- for line in text.split('\n'):
- if line.strip():
- result.append(f"> {line}\n")
- result.append("")
-
- elif tag == 'hr':
- result.append("---\n")
-
- elif tag == 'img':
- alt = element.get('alt', '')
- src = element.get('src', '')
- if src and base_url:
- src = urljoin(base_url, src)
- result.append(f"\n")
-
- elif tag in ['script', 'style', 'noscript', 'meta', 'link', 'head']:
- # Ignore
- pass
-
- # For any others elements, process the children
- else:
- for child in element:
- child_md = element_to_markdown(child, base_url, level + 1)
- if child_md:
- result.extend(child_md)
-
- return result
-
-
-def html_to_markdown(html_content, base_url=None):
- tree = lxml.html.fromstring(html_content)
-
- main_div = tree.xpath('//div[@role="main"]')
- if not main_div:
- main_div = tree.xpath('//main | //div[@class="main"] | //article | //body')
-
- if not main_div:
- return "# Error\n\nMain content not found"
-
- main_div = main_div[0]
-
- markdown_lines = []
- for child in main_div:
- result = element_to_markdown(child, base_url)
- if result:
- markdown_lines.extend(result)
-
- markdown = "".join(markdown_lines)
-
- while "\n\n\n" in markdown:
- markdown = markdown.replace("\n\n\n", "\n\n")
-
- return markdown.strip()
-
-
-def convert_js_to_py_dict(js_text: str) -> dict:
- # Convert javascript dictionary to python dictionary
- js_text = js_text.replace("define(", "").replace(");", "").replace("'", '"')
- js_text = re.sub(r'//.*', '', js_text)
- js_text = re.sub(r'/\*[\s\S]*?\*/', '', js_text)
- js_text = re.sub(r',\s*(?=[}\]])', '', js_text)
- js_text = re.sub(r'([{\[,]\s*)([A-Za-z_][A-Za-z0-9_]*)\s*:', r'\1"\2":', js_text)
- return json.loads(js_text)
+ base_path = os.path.dirname(os.path.abspath(__file__))
+ resources_path = Path(base_path) / 'resources'
+ return resources_path
+
+def get_mcp_icon_uri():
+ name = "app.png"
+ icon_path = get_resources_path().joinpath(name)
+ icon_data = base64.standard_b64encode(icon_path.read_bytes()).decode()
+ return f"data:image/png;base64,{icon_data}"
From a45623927c4161e471fc4c72df4a06eb4cd5b98b Mon Sep 17 00:00:00 2001
From: David <3dgiordano@gmail.com>
Date: Thu, 6 Nov 2025 19:23:01 -0300
Subject: [PATCH 2/4] Skills Tool
---
pyproject.toml | 2 +-
server.py | 2 +
tools/skills_manager.py | 150 +++++++++++++++++++++++++++
tools/skills_utils.py | 225 ++++++++++++++++++++++++++++++++++++++++
4 files changed, 378 insertions(+), 1 deletion(-)
create mode 100644 tools/skills_manager.py
create mode 100644 tools/skills_utils.py
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 8010c8f..92dd9a4 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
def register_tools(mcp, token: Optional[PerfectoToken]):
"""
@@ -18,3 +19,4 @@ 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)
diff --git a/tools/skills_manager.py b/tools/skills_manager.py
new file mode 100644
index 0000000..f9f0106
--- /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 HelpManager
+
+ 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