Expose source management through MCP

Author: httpdss

docs/custom-structures.md

@@ -75,3 +75,15 @@ Existing custom-structure behavior remains compatible:
 4. If none of the above are used, StructKit uses its bundled contrib structures.
 
 When a custom path or named source is selected, StructKit still falls back to bundled contrib structures for `list`, `search`, and `generate` in the same way as the existing `STRUCTKIT_STRUCTURES_PATH` workflow.
+
+### MCP tools
+
+The MCP server exposes the same source-management workflow for AI clients:
+
+- `list_sources`
+- `add_source`
+- `remove_source`
+- `show_source`
+- `validate_source`
+
+The structure tools also accept named sources. Pass `source` to `list_structures`, `get_structure_info`, or `generate_structure`, or use a source-prefixed structure name such as `company/project/python` with `get_structure_info` and `generate_structure`.

structkit/commands/mcp.py

@@ -50,6 +50,11 @@ def execute(self, args):
             print("  - get_structure_info: Get detailed information about a structure")
             print("  - generate_structure: Generate structures with various options")
             print("  - validate_structure: Validate structure configuration files")
+            print("  - list_sources: List configured custom structure sources")
+            print("  - add_source: Add or replace a local custom structure source")
+            print("  - remove_source: Remove a configured custom structure source")
+            print("  - show_source: Show a configured custom structure source")
+            print("  - validate_source: Validate a configured custom structure source")
             print("\nExamples:")
             print("  structkit mcp --server --transport stdio --debug")
             print("  structkit mcp --server --transport http --host 127.0.0.1 --port 9000 --path /mcp --uvicorn-log-level debug")

structkit/mcp_server.py

@@ -6,18 +6,29 @@
 2. Getting detailed information about structures
 3. Generating structures with various options
 4. Validating structure configurations
+5. Managing named custom structure sources
 """
 import asyncio
 import logging
 import os
 import sys
 import yaml
+from types import SimpleNamespace
 from typing import Any, Dict, Optional
 
 from fastmcp import FastMCP
 
 from structkit.commands.generate import GenerateCommand
 from structkit.commands.validate import ValidateCommand
+from structkit.sources import (
+    SourceConfigError,
+    add_source as add_config_source,
+    get_source_path,
+    read_sources,
+    remove_source as remove_config_source,
+    resolve_structures_path,
+    validate_source as validate_config_source,
+)
 from structkit import __version__
 
 
@@ -32,13 +43,20 @@ def __init__(self):
     # =====================
     # Tool logic (transport-agnostic)
     # =====================
-    def _list_structures_logic(self, structures_path: Optional[str] = None) -> str:
+    def _list_structures_logic(self, structures_path: Optional[str] = None, source: Optional[str] = None) -> str:
         this_file = os.path.dirname(os.path.realpath(__file__))
         contribs_path = os.path.join(this_file, "contribs")
 
+        try:
+            effective_structures_path, _ = resolve_structures_path(
+                SimpleNamespace(structures_path=structures_path, source=source)
+            )
+        except SourceConfigError as e:
+            return f"❗ {e}"
+
         paths_to_list = [contribs_path]
-        if structures_path:
-            paths_to_list = [structures_path, contribs_path]
+        if effective_structures_path:
+            paths_to_list = [effective_structures_path, contribs_path]
 
         all_structures = set()
         for path in paths_to_list:
@@ -56,17 +74,28 @@ def _list_structures_logic(self, structures_path: Optional[str] = None) -> str:
         result_text += "\n\nNote: Structures with '+' sign are custom structures"
         return result_text
 
-    def _get_structure_info_logic(self, structure_name: Optional[str], structures_path: Optional[str] = None) -> str:
+    def _get_structure_info_logic(
+        self,
+        structure_name: Optional[str],
+        structures_path: Optional[str] = None,
+        source: Optional[str] = None,
+    ) -> str:
         if not structure_name:
             return "Error: structure_name is required"
 
         # Resolve path
         if structure_name.startswith("file://") and structure_name.endswith(".yaml"):
             file_path = structure_name[7:]
         else:
+            try:
+                effective_structures_path, resolved_structure_name = resolve_structures_path(
+                    SimpleNamespace(structures_path=structures_path, source=source), structure_name
+                )
+            except SourceConfigError as e:
+                return f"❗ {e}"
             this_file = os.path.dirname(os.path.realpath(__file__))
-            base = structures_path or os.path.join(this_file, "contribs")
-            file_path = os.path.join(base, f"{structure_name}.yaml")
+            base = effective_structures_path or os.path.join(this_file, "contribs")
+            file_path = os.path.join(base, f"{resolved_structure_name}.yaml")
 
         if not os.path.exists(file_path):
             return f"❗ Structure not found: {file_path}"
@@ -119,6 +148,7 @@ def _generate_structure_logic(
         dry_run: bool = False,
         mappings: Optional[Dict[str, str]] = None,
         structures_path: Optional[str] = None,
+        source: Optional[str] = None,
     ) -> str:
         class Args:
             pass
@@ -128,6 +158,7 @@ class Args:
         args.output = "console" if output == "console" else "file"
         args.dry_run = dry_run
         args.structures_path = structures_path
+        args.source = source
         args.vars = None
         args.mappings_file = None
         args.backup = None
@@ -167,6 +198,54 @@ class Args:
                 return f"Dry run completed for structure '{structure_definition}' at '{base_path}'"
             return f"Structure '{structure_definition}' generated successfully at '{base_path}'"
 
+    def _list_sources_logic(self) -> str:
+        sources = read_sources()
+        if not sources:
+            return "No sources configured"
+
+        lines = ["📚 Configured structure sources\n"]
+        for name in sorted(sources):
+            lines.append(f" - {name}: {sources[name]}\n")
+        return "".join(lines).rstrip()
+
+    def _add_source_logic(self, name: Optional[str], path_or_url: Optional[str]) -> str:
+        if not name:
+            return "Error: name is required"
+        if not path_or_url:
+            return "Error: path_or_url is required"
+        try:
+            source_path = add_config_source(name, path_or_url)
+        except SourceConfigError as e:
+            return f"❗ {e}"
+        return f"Added source '{name}': {source_path}"
+
+    def _remove_source_logic(self, name: Optional[str]) -> str:
+        if not name:
+            return "Error: name is required"
+        try:
+            source_path = remove_config_source(name)
+        except SourceConfigError as e:
+            return f"❗ {e}"
+        return f"Removed source '{name}': {source_path}"
+
+    def _show_source_logic(self, name: Optional[str]) -> str:
+        if not name:
+            return "Error: name is required"
+        try:
+            source_path = get_source_path(name)
+        except SourceConfigError as e:
+            return f"❗ {e}"
+        return f"{name}: {source_path}"
+
+    def _validate_source_logic(self, name: Optional[str]) -> str:
+        if not name:
+            return "Error: name is required"
+        try:
+            source_path = validate_config_source(name)
+        except SourceConfigError as e:
+            return f"❗ {e}"
+        return f"Source '{name}' is valid: {source_path}"
+
     def _validate_structure_logic(self, yaml_file: Optional[str]) -> str:
         if not yaml_file:
             return "Error: yaml_file is required"
@@ -198,19 +277,23 @@ class Args:
     # =====================
     def _register_tools(self):
         @self.app.tool(name="list_structures", description="List all available structure definitions")
-        async def list_structures(structures_path: Optional[str] = None) -> str:
-            self.logger.debug(f"MCP request: list_structures args={{'structures_path': {structures_path!r}}}")
-            result = self._list_structures_logic(structures_path)
+        async def list_structures(structures_path: Optional[str] = None, source: Optional[str] = None) -> str:
+            self.logger.debug(
+                f"MCP request: list_structures args={{'structures_path': {structures_path!r}, 'source': {source!r}}}"
+            )
+            result = self._list_structures_logic(structures_path, source)
             preview = result if len(result) <= 1000 else result[:1000] + f"... [truncated {len(result)-1000} chars]"
             self.logger.debug(f"MCP response: list_structures len={len(result)} preview=\n{preview}")
             return result
 
         @self.app.tool(name="get_structure_info", description="Get detailed information about a specific structure")
-        async def get_structure_info(structure_name: str, structures_path: Optional[str] = None) -> str:
+        async def get_structure_info(
+            structure_name: str, structures_path: Optional[str] = None, source: Optional[str] = None
+        ) -> str:
             self.logger.debug(
-                f"MCP request: get_structure_info args={{'structure_name': {structure_name!r}, 'structures_path': {structures_path!r}}}"
+                f"MCP request: get_structure_info args={{'structure_name': {structure_name!r}, 'structures_path': {structures_path!r}, 'source': {source!r}}}"
             )
-            result = self._get_structure_info_logic(structure_name, structures_path)
+            result = self._get_structure_info_logic(structure_name, structures_path, source)
             preview = result if len(result) <= 1000 else result[:1000] + f"... [truncated {len(result)-1000} chars]"
             self.logger.debug(f"MCP response: get_structure_info len={len(result)} preview=\n{preview}")
             return result
@@ -223,6 +306,7 @@ async def generate_structure(
             dry_run: bool = False,
             mappings: Optional[Dict[str, str]] = None,
             structures_path: Optional[str] = None,
+            source: Optional[str] = None,
         ) -> str:
             self.logger.debug(
                 "MCP request: generate_structure args=%s",
@@ -233,6 +317,7 @@ async def generate_structure(
                     "dry_run": dry_run,
                     "mappings": mappings,
                     "structures_path": structures_path,
+                    "source": source,
                 },
             )
             result = self._generate_structure_logic(
@@ -242,11 +327,52 @@ async def generate_structure(
                 dry_run,
                 mappings,
                 structures_path,
+                source,
             )
             preview = result if len(result) <= 1000 else result[:1000] + f"... [truncated {len(result)-1000} chars]"
             self.logger.debug(f"MCP response: generate_structure len={len(result)} preview=\n{preview}")
             return result
 
+        @self.app.tool(name="list_sources", description="List configured custom structure sources")
+        async def list_sources() -> str:
+            self.logger.debug("MCP request: list_sources args={}")
+            result = self._list_sources_logic()
+            preview = result if len(result) <= 1000 else result[:1000] + f"... [truncated {len(result)-1000} chars]"
+            self.logger.debug(f"MCP response: list_sources len={len(result)} preview=\n{preview}")
+            return result
+
+        @self.app.tool(name="add_source", description="Add or replace a local custom structure source")
+        async def add_source(name: str, path_or_url: str) -> str:
+            self.logger.debug(f"MCP request: add_source args={{'name': {name!r}, 'path_or_url': {path_or_url!r}}}")
+            result = self._add_source_logic(name, path_or_url)
+            preview = result if len(result) <= 1000 else result[:1000] + f"... [truncated {len(result)-1000} chars]"
+            self.logger.debug(f"MCP response: add_source len={len(result)} preview=\n{preview}")
+            return result
+
+        @self.app.tool(name="remove_source", description="Remove a configured custom structure source")
+        async def remove_source(name: str) -> str:
+            self.logger.debug(f"MCP request: remove_source args={{'name': {name!r}}}")
+            result = self._remove_source_logic(name)
+            preview = result if len(result) <= 1000 else result[:1000] + f"... [truncated {len(result)-1000} chars]"
+            self.logger.debug(f"MCP response: remove_source len={len(result)} preview=\n{preview}")
+            return result
+
+        @self.app.tool(name="show_source", description="Show a configured custom structure source")
+        async def show_source(name: str) -> str:
+            self.logger.debug(f"MCP request: show_source args={{'name': {name!r}}}")
+            result = self._show_source_logic(name)
+            preview = result if len(result) <= 1000 else result[:1000] + f"... [truncated {len(result)-1000} chars]"
+            self.logger.debug(f"MCP response: show_source len={len(result)} preview=\n{preview}")
+            return result
+
+        @self.app.tool(name="validate_source", description="Validate a configured custom structure source")
+        async def validate_source(name: str) -> str:
+            self.logger.debug(f"MCP request: validate_source args={{'name': {name!r}}}")
+            result = self._validate_source_logic(name)
+            preview = result if len(result) <= 1000 else result[:1000] + f"... [truncated {len(result)-1000} chars]"
+            self.logger.debug(f"MCP response: validate_source len={len(result)} preview=\n{preview}")
+            return result
+
         @self.app.tool(name="validate_structure", description="Validate a structure configuration YAML file")
         async def validate_structure(yaml_file: str) -> str:
             self.logger.debug(f"MCP request: validate_structure args={{'yaml_file': {yaml_file!r}}}")
@@ -323,8 +449,9 @@ async def _handle_get_structure_info(self, params: Dict[str, Any]):
         """Compatibility method for tests that expect MCP-style responses."""
         structure_name = params.get('structure_name')
         structures_path = params.get('structures_path')
+        source = params.get('source')
 
-        result_text = self._get_structure_info_logic(structure_name, structures_path)
+        result_text = self._get_structure_info_logic(structure_name, structures_path, source)
 
         # Mock MCP response structure
         class MockContent:

tests/test_mcp_integration.py

@@ -43,6 +43,66 @@ def test_generate_structure_logic(self):
             )
             self.assertIsInstance(text, str)
 
+
+    def test_source_management_logic(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            config_path = os.path.join(temp_dir, 'sources.json')
+            source_dir = os.path.join(temp_dir, 'source')
+            os.makedirs(source_dir)
+
+            from unittest.mock import patch
+            with patch.dict(os.environ, {'STRUCTKIT_SOURCES_CONFIG': config_path}):
+                self.assertEqual(self.server._list_sources_logic(), 'No sources configured')
+
+                added = self.server._add_source_logic('company', source_dir)
+                self.assertIn("Added source 'company'", added)
+
+                listed = self.server._list_sources_logic()
+                self.assertIn('company', listed)
+                self.assertIn(source_dir, listed)
+
+                shown = self.server._show_source_logic('company')
+                self.assertIn(source_dir, shown)
+
+                validated = self.server._validate_source_logic('company')
+                self.assertIn("Source 'company' is valid", validated)
+
+                removed = self.server._remove_source_logic('company')
+                self.assertIn("Removed source 'company'", removed)
+
+    def test_source_management_logic_errors(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            config_path = os.path.join(temp_dir, 'sources.json')
+
+            from unittest.mock import patch
+            with patch.dict(os.environ, {'STRUCTKIT_SOURCES_CONFIG': config_path}):
+                self.assertIn('name is required', self.server._add_source_logic(None, temp_dir))
+                self.assertIn('path_or_url is required', self.server._add_source_logic('company', None))
+                self.assertIn('Only local filesystem', self.server._add_source_logic('remote', 'https://example.com/templates.git'))
+                self.assertIn('Unknown source', self.server._show_source_logic('missing'))
+
+    def test_structure_logic_accepts_named_source(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            config_path = os.path.join(temp_dir, 'sources.json')
+            source_dir = os.path.join(temp_dir, 'source')
+            os.makedirs(os.path.join(source_dir, 'project'))
+            structure_file = os.path.join(source_dir, 'project', 'python.yaml')
+            with open(structure_file, 'w') as f:
+                yaml.dump({'description': 'Custom Python structure', 'files': []}, f)
+
+            from unittest.mock import patch
+            with patch.dict(os.environ, {'STRUCTKIT_SOURCES_CONFIG': config_path}):
+                self.server._add_source_logic('company', source_dir)
+
+                listed = self.server._list_structures_logic(source='company')
+                self.assertIn('+ project/python', listed)
+
+                explicit = self.server._get_structure_info_logic('project/python', source='company')
+                self.assertIn('Custom Python structure', explicit)
+
+                prefixed = self.server._get_structure_info_logic('company/project/python')
+                self.assertIn('Custom Python structure', prefixed)
+
     def test_validate_structure_logic(self):
         # Missing yaml_file
         text = self.server._validate_structure_logic(None)