28.4 扩展性设计

约 1672 字大约 6 分钟

28.4.1 扩展性概述

Claude Code 的扩展性设计是其能够适应不同使用场景和需求的关键。良好的扩展性设计使得系统可以：

轻松添加新功能：通过插件机制添加新能力
支持多种集成方式：与不同的开发工具和环境集成
适应不同规模：从小型项目到大型企业级应用
保持向后兼容：新版本不影响现有功能
支持自定义配置：根据具体需求进行定制

28.4.2 插件系统架构

1. 插件接口设计

from abc import ABC, abstractmethod
from typing import Dict, List, Any

class PluginInterface(ABC):
"""插件接口"""

@property
@abstractmethod
def name(self) -> str:
"""插件名称"""
pass

@property
@abstractmethod
def version(self) -> str:
"""插件版本"""
pass

@property
@abstractmethod
def description(self) -> str:
"""插件描述"""
pass

@abstractmethod
def initialize(self, context: PluginContext):
"""初始化插件"""
pass

@abstractmethod
def execute(self, request: PluginRequest) -> PluginResponse:
"""执行插件功能"""
pass

@abstractmethod
def cleanup(self):
"""清理插件资源"""
pass

def get_capabilities(self) -> List[str]:
"""获取插件能力"""
return []

def validate_config(self, config: Dict[str, Any]) -> bool:
"""验证配置"""
return True

2. 插件管理器

import importlib.util
from typing import Dict, List, Any
from loguru import logger

class PluginManager:
    """插件管理器"""

    def __init__(self):
        self.plugins: Dict[str, PluginInterface] = {}
        self.plugin_configs: Dict[str, Dict[str, Any]] = {}
        self.plugin_dependencies: Dict[str, List[str]] = {}
        self.load_order: List[str] = []

    def register_plugin(self, plugin: PluginInterface,
                       config: Dict[str, Any] = None):
        """注册插件"""
        plugin_name = plugin.name

        # 检查是否已注册
        if plugin_name in self.plugins:
            logger.warning(f"Plugin already registered: {plugin_name}")
            return

        # 验证配置
        if config and not plugin.validate_config(config):
            raise ValueError(f"Invalid config for plugin: {plugin_name}")

        # 存储插件
        self.plugins[plugin_name] = plugin
        self.plugin_configs[plugin_name] = config or {}

        logger.info(f"Plugin registered: {plugin_name} v{plugin.version}")

    def load_plugin(self, plugin_path: str, config: Dict[str, Any] = None):
        """加载插件"""
        # 动态导入插件模块
        spec = importlib.util.spec_from_file_location("plugin", plugin_path)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)

        # 获取插件类
        plugin_class = getattr(module, 'Plugin')

        # 实例化插件
        plugin = plugin_class()

        # 注册插件
        self.register_plugin(plugin, config)

    def initialize_plugins(self):
        """初始化所有插件"""
        # 解析依赖关系
        self._resolve_dependencies()

        # 按顺序初始化插件
        for plugin_name in self.load_order:
            plugin = self.plugins[plugin_name]
            config = self.plugin_configs[plugin_name]

            context = PluginContext(
                plugin_manager=self,
                config=config
            )

            try:
                plugin.initialize(context)
                logger.info(f"Plugin initialized: {plugin_name}")
            except Exception as e:
                logger.error(f"Failed to initialize plugin {plugin_name}: {e}")
                raise

    def execute_plugin(self, plugin_name: str,
                       request: PluginRequest) -> PluginResponse:
        """执行插件"""
        plugin = self.plugins.get(plugin_name)

        if not plugin:
            raise ValueError(f"Plugin not found: {plugin_name}")

        try:
            response = plugin.execute(request)
            return response
        except Exception as e:
            logger.error(f"Error executing plugin {plugin_name}: {e}")
            raise

    def cleanup_plugins(self):
        """清理所有插件"""
        for plugin_name in reversed(self.load_order):
            plugin = self.plugins[plugin_name]
            try:
                plugin.cleanup()
                logger.info(f"Plugin cleaned up: {plugin_name}")
            except Exception as e:
                logger.error(f"Error cleaning up plugin {plugin_name}: {e}")

    def _resolve_dependencies(self):
        """解析插件依赖关系"""
        # 构建依赖图
        graph = {}
        for plugin_name, deps in self.plugin_dependencies.items():
            graph[plugin_name] = deps

        # 拓扑排序
        self.load_order = self._topological_sort(graph)

    def _topological_sort(self, graph: Dict[str, List[str]]) -> List[str]:
        """拓扑排序"""
        visited = set()
        result = []

        def visit(node):
            if node in visited:
                return

            visited.add(node)

            for dep in graph.get(node, []):
                visit(dep)

            result.append(node)

        for node in graph:
            visit(node)

        return result

3. 插件上下文

class PluginContext:
"""插件上下文"""
def __init__(self, plugin_manager: PluginManager,
config: Dict[str, Any]):
self.plugin_manager = plugin_manager
self.config = config
self.shared_data: Dict[str, Any] = {}
self.event_bus: EventBus = EventBus()
def get_config(self, key: str, default: Any = None) -> Any:
"""获取配置"""
return self.config.get(key, default)
def set_shared_data(self, key: str, value: Any):
"""设置共享数据"""
self.shared_data[key] = value
def get_shared_data(self, key: str, default: Any = None) -> Any:
"""获取共享数据"""
return self.shared_data.get(key, default)
def emit_event(self, event_name: str, data: Dict[str, Any]):
"""发送事件"""
self.event_bus.emit(event_name, data)
def on_event(self, event_name: str, handler: callable):
"""监听事件"""
self.event_bus.on(event_name, handler)

28.4.3 工具扩展机制

1. 工具注册

class ToolRegistry:
    """工具注册表"""

    def __init__(self):
        self.tools: Dict[str, Tool] = {}
        self.tool_categories: Dict[str, List[str]] = {}

    def register_tool(self, tool: Tool, category: str = None):
        """注册工具"""
        tool_id = tool.id

        if tool_id in self.tools:
            logger.warning(f"Tool already registered: {tool_id}")
            return

        self.tools[tool_id] = tool

        if category:
            if category not in self.tool_categories:
                self.tool_categories[category] = []
            self.tool_categories[category].append(tool_id)

        logger.info(f"Tool registered: {tool_id}")

    def get_tool(self, tool_id: str) -> Tool:
        """获取工具"""
        return self.tools.get(tool_id)

    def get_tools_by_category(self, category: str) -> List[Tool]:
        """按类别获取工具"""
        tool_ids = self.tool_categories.get(category, [])
        return [self.tools[tid] for tid in tool_ids]

    def list_tools(self) -> List[str]:
        """列出所有工具"""
        return list(self.tools.keys())

2. 自定义工具

class CustomTool(Tool):
"""自定义工具"""
def __init__(self, tool_id: str, name: str,
description: str, handler: callable):
super().__init__(tool_id, name, description)
self.handler = handler
def execute(self, parameters: Dict[str, Any]) -> ToolResult:
"""执行工具"""
try:
result = self.handler(parameters)
return ToolResult(
success=True,
data=result,
message="Execution successful"
)
except Exception as e:
return ToolResult(
success=False,
error=str(e),
message="Execution failed"
)
# 使用示例
def custom_handler(params: Dict[str, Any]) -> Any:
"""自定义处理函数"""
# 实现自定义逻辑
return {"result": "custom result"}
custom_tool = CustomTool(
tool_id="custom_tool",
name="Custom Tool",
description="A custom tool for specific tasks",
handler=custom_handler
)
registry = ToolRegistry()
registry.register_tool(custom_tool, category="custom")

28.4.4 集成扩展

1. IDE 集成接口

from abc import ABC, abstractmethod

class IDEIntegration(ABC):
    """IDE 集成接口"""

    @abstractmethod
    def get_current_file(self) -> str:
        """获取当前文件"""
        pass

    @abstractmethod
    def get_selection(self) -> str:
        """获取选中的文本"""
        pass

    @abstractmethod
    def open_file(self, file_path: str, line: int = None):
        """打开文件"""
        pass

    @abstractmethod
    def insert_text(self, text: str, position: int = None):
        """插入文本"""
        pass

    @abstractmethod
    def show_notification(self, message: str, level: str = "info"):
        """显示通知"""
        pass

2. VS Code 集成

class VSCodeIntegration(IDEIntegration):
"""VS Code 集成"""
def __init__(self):
self.extension_api = None
def connect(self, extension_api):
"""连接到 VS Code 扩展 API"""
self.extension_api = extension_api
def get_current_file(self) -> str:
"""获取当前文件"""
if not self.extension_api:
raise RuntimeError("Not connected to VS Code")
return self.extension_api.window.activeTextEditor.document.uri.fsPath
def get_selection(self) -> str:
"""获取选中的文本"""
if not self.extension_api:
raise RuntimeError("Not connected to VS Code")
editor = self.extension_api.window.activeTextEditor
selection = editor.selection
return editor.document.getText(selection)
def open_file(self, file_path: str, line: int = None):
"""打开文件"""
if not self.extension_api:
raise RuntimeError("Not connected to VS Code")
uri = self.extension_api.Uri.file(file_path)
self.extension_api.commands.executeCommand(
'vscode.open',
uri,
{
'selection': self.extension_api.Range(
self.extension_api.Position(line or 0, 0),
self.extension_api.Position(line or 0, 0)
)
}
)
def insert_text(self, text: str, position: int = None):
"""插入文本"""
if not self.extension_api:
raise RuntimeError("Not connected to VS Code")
editor = self.extension_api.window.activeTextEditor
edit_position = editor.selection.active
editor.edit(lambda editBuilder: editBuilder.insert(edit_position, text))
def show_notification(self, message: str, level: str = "info"):
"""显示通知"""
if not self.extension_api:
raise RuntimeError("Not connected to VS Code")
self.extension_api.window.showInformationMessage(message)

28.4.5 配置扩展

1. 配置系统

class ConfigurationSystem:
    """配置系统"""

    def __init__(self):
        self.configs: Dict[str, Any] = {}
        self.config_sources: List[ConfigSource] = []
        self.validators: Dict[str, List[callable]] = {}

    def add_source(self, source: ConfigSource):
        """添加配置源"""
        self.config_sources.append(source)

    def add_validator(self, key: str, validator: callable):
        """添加验证器"""
        if key not in self.validators:
            self.validators[key] = []
        self.validators[key].append(validator)

    def load(self):
        """加载配置"""
        for source in self.config_sources:
            config = source.load()
            self._merge_config(config)

    def get(self, key: str, default: Any = None) -> Any:
        """获取配置"""
        value = self._get_nested_value(key)

        if value is None:
            return default

        # 验证值
        validators = self.validators.get(key, [])
        for validator in validators:
            if not validator(value):
                raise ValueError(f"Invalid value for key: {key}")

        return value

    def set(self, key: str, value: Any):
        """设置配置"""
        self._set_nested_value(key, value)

    def _merge_config(self, config: Dict[str, Any]):
        """合并配置"""
        for key, value in config.items():
            if key in self.configs and isinstance(self.configs[key], dict) and isinstance(value, dict):
                self.configs[key].update(value)
            else:
                self.configs[key] = value

    def _get_nested_value(self, key: str) -> Any:
        """获取嵌套值"""
        keys = key.split('.')
        value = self.configs

        for k in keys:
            if isinstance(value, dict) and k in value:
                value = value[k]
            else:
                return None

        return value

    def _set_nested_value(self, key: str, value: Any):
        """设置嵌套值"""
        keys = key.split('.')
        config = self.configs

        for k in keys[:-1]:
            if k not in config:
                config[k] = {}
            config = config[k]

        config[keys[-1]] = value

2. 配置源

from abc import ABC, abstractmethod

class ConfigSource(ABC):
"""配置源"""
@abstractmethod
def load(self) -> Dict[str, Any]:
"""加载配置"""
pass
class FileConfigSource(ConfigSource):
"""文件配置源"""
def __init__(self, file_path: str):
self.file_path = file_path
def load(self) -> Dict[str, Any]:
"""从文件加载配置"""
with open(self.file_path, 'r') as f:
return json.load(f)
class EnvironmentConfigSource(ConfigSource):
"""环境变量配置源"""
def __init__(self, prefix: str = "CLAUDE_"):
self.prefix = prefix
def load(self) -> Dict[str, Any]:
"""从环境变量加载配置"""
config = {}
for key, value in os.environ.items():
if key.startswith(self.prefix):
config_key = key[len(self.prefix):].lower()
config[config_key] = value
return config