28.4 擴充套件性設計
28.4.1 擴充套件性概述
Claude Code 的擴充套件性設計是其能夠適應不同使用場景和需求的關鍵。良好的擴充套件性設計使得系統可以：
- 輕鬆新增新功能 ：透過外掛機制新增新能力
- 支援多種整合方式 ：與不同的開發工具和環境整合
- 適應不同規模 ：從小型專案到大型企業級應用
- 保持向後相容 ：新版本不影響現有功能
- 支援自定義配置 ：根據具體需求進行定製
python
## 28.4.2 插件系统架构
### 1\. 插件接口设计
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
python
def get_capabilities(self) -> List[str]: """获取插件能力""" return []
def validate_config(self, config: Dict[str, Any]) -> bool: """验证配置""" return True
### 2\. 插件管理器
bash
python
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\. 工具注册
bash
python
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 集成接口
bash
python
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\. 配置系统
bash
python
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. 配置源
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