feat: add MCP tools for SOP discovery and retrieval

python/pyproject.toml

@@ -50,10 +50,11 @@ include = [
     "strands_agents_sops/__init__.py",
     "strands_agents_sops/__main__.py",
     "strands_agents_sops/cursor.py",
-    "strands_agents_sops/mcp.py",
     "strands_agents_sops/skills.py",
     "strands_agents_sops/rules.py",
     "strands_agents_sops/utils.py",
+    "strands_agents_sops/mcp/__init__.py",
+    "strands_agents_sops/mcp/server.py",
     "strands_agents_sops/sops/*.md",
     "strands_agents_sops/rules/*.md"
 ]

python/strands_agents_sops/__init__.py

@@ -1,26 +1,17 @@
-from pathlib import Path
-
 from .rules import get_sop_format as get_sop_format
+from .utils import load_builtin_sops
 
 # Load all SOP files as module attributes
-_sops_dir = Path(__file__).parent / "sops"
-
-for _md_file in _sops_dir.glob("*.sop.md"):
-    if _md_file.is_file():
-        # Convert filename to valid Python identifier
-        _attr_name = (
-            _md_file.stem.removesuffix(".sop").replace("-", "_").replace(".", "_")
-        )
-        _sop_name = _md_file.stem.removesuffix(".sop")
-        _content = _md_file.read_text(encoding="utf-8")
+for _sop in load_builtin_sops():
+    _attr_name = _sop["name"].replace("-", "_").replace(".", "_")
 
-        # Load file content as module attribute
-        globals()[_attr_name] = _content
+    # Load file content as module attribute
+    globals()[_attr_name] = _sop["content"]
 
-        # Create wrapper function for each SOP
-        def _make_wrapper(content, name):
-            def wrapper(user_input: str = "") -> str:
-                return f"""<agent-sop name="{name}">
+    # Create wrapper function for each SOP
+    def _make_wrapper(content, name):
+        def wrapper(user_input: str = "") -> str:
+            return f"""<agent-sop name="{name}">
 <content>
 {content}
 </content>
@@ -29,9 +20,6 @@ def wrapper(user_input: str = "") -> str:
 </user-input>
 </agent-sop>"""
 
-            return wrapper
-
-        globals()[f"{_attr_name}_with_input"] = _make_wrapper(_content, _sop_name)
+        return wrapper
 
-# Clean up temporary variables
-del _sops_dir
+    globals()[f"{_attr_name}_with_input"] = _make_wrapper(_sop["content"], _sop["name"])

python/strands_agents_sops/__main__.py

@@ -1,7 +1,7 @@
 import argparse
 
 from .cursor import generate_cursor_commands
-from .mcp import run_mcp_server
+from .mcp.server import AgentSOPMCPServer
 from .rules import output_rules
 from .skills import generate_anthropic_skills
 
@@ -85,7 +85,8 @@ def main():
     else:
         # Default to MCP server
         sop_paths = getattr(args, "sop_paths", None)
-        run_mcp_server(sop_paths=sop_paths)
+        mcp_server = AgentSOPMCPServer(sop_paths=sop_paths)
+        mcp_server.run()
 
 
 if __name__ == "__main__":

python/strands_agents_sops/mcp/__init__.py

@@ -0,0 +1,5 @@
+"""MCP server implementation for Agent SOPs."""
+
+from .server import AgentSOPMCPServer
+
+__all__ = ["AgentSOPMCPServer"]

python/strands_agents_sops/mcp/server.py

@@ -0,0 +1,114 @@
+"""MCP server orchestrator for Agent SOPs."""
+
+import logging
+from collections.abc import Callable
+
+from mcp.server.fastmcp import FastMCP
+
+from ..utils import load_sops
+
+logger = logging.getLogger(__name__)
+
+
+class AgentSOPMCPServer:
+    """MCP server for serving Agent SOPs as prompts and tools."""
+
+    def __init__(self, sop_paths: str | None = None):
+        """Initialize the MCP server.
+
+        Args:
+            sop_paths: Optional colon-separated string of external SOP directory paths
+        """
+        self.sop_paths = sop_paths
+        self.mcp = FastMCP("agent-sop-prompt-server")
+        self.sops = {sop["name"]: sop for sop in load_sops(self.sop_paths)}
+        self._register_sop_prompts(self.sops)
+        self._register_sop_tools()
+
+    def _register_sop_tools(self) -> None:
+        """Register SOP management tools with MCP server."""
+
+        @self.mcp.tool()
+        def list_agent_sops() -> list[dict]:
+            """List all available agent SOPs with name and description.
+
+            Returns:
+                List of SOP dictionaries containing name and description
+            """
+            result = []
+            for sop in self.sops.values():
+                result.append(
+                    {
+                        "name": sop["name"],
+                        "description": sop["description"],
+                    }
+                )
+            return result
+
+        @self.mcp.tool()
+        def get_agent_sop(name: str) -> dict:
+            """Get the full content of a specific SOP.
+
+            Args:
+                name: Name of the SOP to retrieve
+
+            Returns:
+                SOP dictionary with name, description, and full content
+
+            Raises:
+                ValueError: If SOP name is not found
+            """
+            if name in self.sops:
+                sop = self.sops[name]
+                return {
+                    "name": sop["name"],
+                    "description": sop["description"],
+                    "content": sop["content"],
+                }
+
+            raise ValueError(
+                f"SOP '{name}' not found. Available SOPs: {list(self.sops.keys())}"
+            )
+
+    def run(self) -> None:
+        """Start the MCP server."""
+        self.mcp.run()
+
+    def _register_sop_prompts(self, sops: dict[str, dict]) -> None:
+        """Register SOP prompts with MCP server.
+
+        Args:
+            sops: Dictionary of SOP name to SOP dict with name, description, and content
+        """
+        for sop in sops.values():
+            try:
+                handler = self._create_prompt_handler(sop["name"], sop["content"])
+                self.mcp.prompt(name=sop["name"], description=sop["description"])(handler)
+            except Exception as e:
+                raise RuntimeError(
+                    f"Error registering prompt for SOP '{sop['name']}': {e}"
+                ) from e
+
+    def _create_prompt_handler(self, sop_name: str, sop_content: str) -> Callable[[str], str]:
+        """Create a prompt handler for a specific SOP.
+
+        Args:
+            sop_name: Name of the SOP
+            sop_content: Content of the SOP
+
+        Returns:
+            Prompt handler function
+        """
+
+        def get_prompt(user_input: str = "") -> str:
+            return f"""Run this SOP:
+<agent-sop name="{sop_name}">
+<content>
+{sop_content}
+</content>
+<user-input>
+{user_input}
+</user-input>
+</agent-sop>"""
+
+        return get_prompt

python/strands_agents_sops/utils.py

@@ -31,6 +31,39 @@ def expand_sop_paths(sop_paths_str: str) -> list[Path]:
     return paths
 
 
+def _load_sop_file(sop_file: Path) -> dict[str, str] | None:
+    """Load a single SOP file and extract its metadata.
+
+    Args:
+        sop_file: Path to a .sop.md file
+
+    Returns:
+        SOP dictionary with name, content, description, or None if invalid
+    """
+    try:
+        sop_content = sop_file.read_text(encoding="utf-8")
+
+        overview_match = re.search(
+            r"## Overview\s*\n(.*?)(?=\n##|\n#|\Z)", sop_content, re.DOTALL
+        )
+        if not overview_match:
+            logger.warning(f"No Overview section found in {sop_file}")
+            return None
+
+        description = overview_match.group(1).strip().replace("\n", " ")
+        sop_name = sop_file.stem.removesuffix(".sop")
+
+        return {
+            "name": sop_name,
+            "content": sop_content,
+            "description": description,
+        }
+
+    except Exception as e:
+        logger.error(f"Error loading SOP from {sop_file}: {e}")
+        return None
+
+
 def load_external_sops(sop_directories: list[Path]) -> list[dict[str, Any]]:
     """Load SOPs from external directories.
 
@@ -56,34 +89,56 @@ def load_external_sops(sop_directories: list[Path]) -> list[dict[str, Any]]:
                 if not sop_file.is_file():
                     continue
 
-                try:
-                    sop_content = sop_file.read_text(encoding="utf-8")
-
-                    # Extract overview section for description
-                    overview_match = re.search(
-                        r"## Overview\s*\n(.*?)(?=\n##|\n#|\Z)", sop_content, re.DOTALL
-                    )
-                    if not overview_match:
-                        logger.warning(f"No Overview section found in {sop_file}")
-                        continue
-
-                    description = overview_match.group(1).strip().replace("\n", " ")
-                    sop_name = sop_file.stem.removesuffix(".sop")
-
-                    external_sops.append(
-                        {
-                            "name": sop_name,
-                            "content": sop_content,
-                            "description": description,
-                        }
-                    )
-
-                except Exception as e:
-                    logger.error(f"Error loading SOP from {sop_file}: {e}")
-                    continue
+                sop = _load_sop_file(sop_file)
+                if sop:
+                    external_sops.append(sop)
 
         except Exception as e:
             logger.error(f"Error scanning directory {directory}: {e}")
             continue
 
     return external_sops
+
+
+def load_builtin_sops() -> list[dict[str, str]]:
+    """Load all built-in SOPs from the sops directory.
+
+    Returns:
+        List of SOP dictionaries with name, content, description
+    """
+    sops = []
+    sops_dir = Path(__file__).parent / "sops"
+
+    if not sops_dir.exists():
+        logger.warning(f"Built-in SOPs directory not found: {sops_dir}")
+        return sops
+
+    for md_file in sops_dir.glob("*.sop.md"):
+        if md_file.is_file():
+            sop = _load_sop_file(md_file)
+            if sop:
+                sops.append(sop)
+
+    return sops
+
+
+def load_sops(external_sop_paths: str | None = None) -> list[dict[str, Any]]:
+    """Load all SOPs from external paths and built-in directory.
+
+    Args:
+        external_sop_paths: Optional colon-separated string of external SOP directory paths
+
+    Returns:
+        List of SOP dictionaries with name, content, description (external SOPs first)
+    """
+    external_directories = expand_sop_paths(external_sop_paths) if external_sop_paths else []
+    sops = load_external_sops(external_directories) + load_builtin_sops()
+
+    all_sops = []
+    seen_names = set()
+    for sop in sops:
+        if sop["name"] not in seen_names:
+            seen_names.add(sop["name"])
+            all_sops.append(sop)
+
+    return all_sops