Merge pull request #2 from jhbarnett/feat/upstream-cherry-picks

feat: auto-discover and load CLI-installed plugins into SDK sessions

Author: jhbarnett

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "claude-code-telegram"
-version = "1.5.0"
+version = "1.6.1"
 description = "Telegram bot for remote Claude Code access with comprehensive configuration management"
 readme = "README.md"
 license = "MIT"

src/claude/sdk_integration.py

@@ -1,6 +1,7 @@
 """Claude Code Python SDK integration."""
 
 import asyncio
+import json
 import os
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -196,6 +197,12 @@ def _stderr_callback(line: str) -> None:
                 sdk_allowed_tools = self.config.claude_allowed_tools
                 sdk_disallowed_tools = self.config.claude_disallowed_tools
 
+            # Discover installed CLI plugins
+            discovered_plugins = self._discover_plugins()
+            extra_plugin_dirs = self.config.claude_plugin_dirs or []
+            for d in extra_plugin_dirs:
+                discovered_plugins.append({"type": "local", "path": d})
+
             # Build Claude Agent options
             options = ClaudeAgentOptions(
                 max_turns=self.config.claude_max_turns,
@@ -212,7 +219,8 @@ def _stderr_callback(line: str) -> None:
                     "excludedCommands": self.config.sandbox_excluded_commands or [],
                 },
                 system_prompt=base_prompt,
-                setting_sources=["project"],
+                setting_sources=["user", "project"],
+                plugins=discovered_plugins if discovered_plugins else None,
                 stderr=_stderr_callback,
             )
 
@@ -525,6 +533,68 @@ async def _handle_stream_message(
         except Exception as e:
             logger.warning("Stream callback failed", error=str(e))
 
+    def _discover_plugins(self) -> List[Dict[str, str]]:
+        """Discover installed CLI plugins from ~/.claude/plugins/cache/.
+
+        Returns a list of plugin descriptors suitable for ClaudeAgentOptions.plugins,
+        e.g. [{"type": "local", "path": "/root/.claude/plugins/cache/mp/plug/1.0"}].
+        """
+        plugins: List[Dict[str, str]] = []
+        cache_dir = Path.home() / ".claude" / "plugins" / "cache"
+        if not cache_dir.is_dir():
+            return plugins
+
+        # Also consult installed_plugins.json for the pinned version of each plugin,
+        # so we resolve the correct version directory under cache/.
+        installed_versions: Dict[str, str] = {}
+        installed_json = Path.home() / ".claude" / "plugins" / "installed_plugins.json"
+        if installed_json.is_file():
+            try:
+                data = json.loads(installed_json.read_text(encoding="utf-8"))
+                for entry in data if isinstance(data, list) else []:
+                    name = entry.get("name", "")
+                    version = entry.get("version", "")
+                    if name and version:
+                        installed_versions[name] = version
+            except (json.JSONDecodeError, OSError) as e:
+                logger.debug("Could not read installed_plugins.json", error=str(e))
+
+        for marketplace_dir in cache_dir.iterdir():
+            if not marketplace_dir.is_dir():
+                continue
+            for plugin_dir in marketplace_dir.iterdir():
+                if not plugin_dir.is_dir():
+                    continue
+                # Determine which version to use: prefer installed_plugins.json,
+                # then fall back to the lexicographically latest version dir.
+                lookup_key = f"{plugin_dir.name}@{marketplace_dir.name}"
+                pinned = installed_versions.get(lookup_key)
+                version_dir = None
+                if pinned:
+                    candidate = plugin_dir / pinned
+                    if candidate.is_dir():
+                        version_dir = candidate
+                if version_dir is None:
+                    version_dirs = sorted(
+                        [d for d in plugin_dir.iterdir() if d.is_dir()],
+                        key=lambda d: d.name,
+                        reverse=True,
+                    )
+                    if version_dirs:
+                        version_dir = version_dirs[0]
+                if version_dir and (
+                    version_dir / ".claude-plugin" / "plugin.json"
+                ).is_file():
+                    plugins.append({"type": "local", "path": str(version_dir)})
+                    logger.info(
+                        "Discovered plugin",
+                        plugin=plugin_dir.name,
+                        marketplace=marketplace_dir.name,
+                        path=str(version_dir),
+                    )
+
+        return plugins
+
     def _load_mcp_config(self, config_path: Path) -> Dict[str, Any]:
         """Load MCP server configuration from a JSON file.
 

src/config/settings.py

@@ -120,6 +120,10 @@ class Settings(BaseSettings):
         default=[],
         description="List of explicitly disallowed Claude tools/commands",
     )
+    claude_plugin_dirs: Optional[List[str]] = Field(
+        default=None,
+        description="Additional plugin directories to load (comma-separated paths)",
+    )
 
     # Sandbox settings
     sandbox_enabled: bool = Field(
@@ -302,6 +306,19 @@ def parse_int_list(cls, v: Any) -> Optional[List[int]]:
             return [int(uid) for uid in v]
         return v  # type: ignore[no-any-return]
 
+    @field_validator("claude_plugin_dirs", mode="before")
+    @classmethod
+    def parse_claude_plugin_dirs(cls, v: Any) -> Optional[List[str]]:
+        """Parse comma-separated plugin directory paths."""
+        if v is None:
+            return None
+        if isinstance(v, str):
+            dirs = [d.strip() for d in v.split(",") if d.strip()]
+            return dirs if dirs else None
+        if isinstance(v, list):
+            return [str(d) for d in v]
+        return v  # type: ignore[no-any-return]
+
     @field_validator("claude_allowed_tools", mode="before")
     @classmethod
     def parse_claude_allowed_tools(cls, v: Any) -> Optional[List[str]]: