# 工具调用参数
{
    "tool_name": "code_interpreter",
    "arguments": {
        "code": "print('Hello, World!')",
        "language": "python"
    }
}

# 工具执行结果
Union[str, List[ContentItem], dict]

# ContentItem 结构
ContentItem(
    text="执行结果文本",
    image="base64编码的图片",
    file="文件路径"
)

graph TB
    subgraph "工具抽象层"
        BaseTool[BaseTool<br/>工具基类]
        ToolRegistry[TOOL_REGISTRY<br/>全局注册表]
        RegisterDecorator[@register_tool<br/>注册装饰器]
        ToolFactory[get_tool<br/>工具工厂]
        BaseToolWithFileAccess[BaseToolWithFileAccess<br/>文件访问工具基类]
    end
    
    subgraph "核心工具集"
        CodeInterpreter[CodeInterpreter<br/>代码解释器]
        WebSearch[WebSearch<br/>网络搜索]
        DocParser[DocParser<br/>文档解析]
        ImageGen[ImageGen<br/>图像生成]
        Retrieval[Retrieval<br/>RAG检索]
        AmapWeather[AmapWeather<br/>天气查询]
    end
    
    subgraph "扩展工具集"
        SimpleDocParser[SimpleDocParser<br/>简单文档解析]
        WebExtractor[WebExtractor<br/>网页提取]
        Storage[Storage<br/>存储管理]
        ImageSearch[ImageSearch<br/>图像搜索]
        PythonExecutor[PythonExecutor<br/>Python执行器]
        MCPManager[MCPManager<br/>MCP工具管理]
    end
    
    subgraph "搜索工具集"
        KeywordSearch[KeywordSearch<br/>关键词搜索]
        VectorSearch[VectorSearch<br/>向量搜索]
        HybridSearch[HybridSearch<br/>混合搜索]
        FrontPageSearch[FrontPageSearch<br/>首页搜索]
    end
    
    subgraph "工具管理"
        ToolValidator[工具验证器]
        ParameterValidator[参数验证器]
    end
    
    subgraph "执行环境"
        JupyterKernel[Jupyter内核<br/>代码执行]
        SearchEngine[搜索引擎<br/>Bing/Google]
        DocumentParsers[文档解析器<br/>PDF/Word/PPT]
        ImageServices[图像服务<br/>在线API]
        VectorStore[向量存储<br/>文档检索]
        ProcessManager[进程管理器]
        FileSystem[文件系统]
        HTTPClient[HTTP 客户端]
    end
    
    subgraph "搜索工具子系统"
        Retrieval --> HybridSearch
        HybridSearch --> VectorSearch
        HybridSearch --> KeywordSearch
        HybridSearch --> FrontPageSearch
    end
    
    %% 注册关系
    RegisterDecorator --> ToolRegistry
    BaseTool --> ToolRegistry
    ToolFactory --> ToolRegistry
    ToolFactory --> ToolValidator
    ToolValidator --> ParameterValidator
    
    %% 工具实现
    BaseTool --> BaseToolWithFileAccess
    BaseToolWithFileAccess --> CodeInterpreter
    BaseToolWithFileAccess --> DocParser
    BaseToolWithFileAccess --> Retrieval
    
    BaseTool --> WebSearch
    BaseTool --> ImageGen
    BaseTool --> AmapWeather
    BaseTool --> SimpleDocParser
    BaseTool --> WebExtractor
    BaseTool --> Storage
    BaseTool --> ImageSearch
    BaseTool --> PythonExecutor
    BaseTool --> MCPManager
    BaseTool --> KeywordSearch
    BaseTool --> VectorSearch
    BaseTool --> HybridSearch
    BaseTool --> FrontPageSearch
    
    %% 执行依赖
    CodeInterpreter --> JupyterKernel
    CodeInterpreter --> ProcessManager
    DocParser --> FileSystem
    WebSearch --> HTTPClient
    WebSearch --> SearchEngine
    DocParser --> DocumentParsers
    ImageGen --> ImageServices
    Retrieval --> VectorStore

sequenceDiagram
    participant A as Agent
    participant T as Tool
    participant V as Validator
    participant E as Executor
    participant R as Resource
    
    A->>T: call(params, files, **kwargs)
    T->>V: _verify_json_format_args(params)
    V->>V: 参数格式验证
    V->>V: 必需参数检查
    V-->>T: 验证通过
    
    T->>E: 执行具体工具逻辑
    E->>R: 访问外部资源
    R-->>E: 返回资源数据
    E->>E: 处理和格式化结果
    E-->>T: 返回执行结果
    
    T->>T: 结果后处理
    T-->>A: 返回最终结果

sequenceDiagram
    participant A as Agent
    participant CI as CodeInterpreter
    participant KM as KernelManager
    participant JK as Jupyter Kernel
    participant FS as FileSystem
    
    A->>CI: call(code, files, timeout)
    CI->>CI: 解析代码参数
    CI->>FS: 复制文件到工作目录
    
    CI->>KM: 获取或创建内核
    alt 内核不存在
        KM->>JK: 启动新的 Jupyter 内核
        JK-->>KM: 内核就绪
        KM->>JK: 执行初始化代码
    end
    
    CI->>JK: 执行用户代码
    JK->>JK: 代码执行
    
    loop 处理执行结果
        JK-->>CI: 返回执行消息
        CI->>CI: 解析消息类型
        alt 标准输出
            CI->>CI: 收集文本输出
        else 图像输出
            CI->>FS: 保存图像到静态目录
            CI->>CI: 生成图像 URL
        else 错误输出
            CI->>CI: 收集错误信息
        end
    end
    
    CI->>CI: 格式化最终结果
    CI-->>A: 返回执行结果

sequenceDiagram
    participant A as Agent
    participant R as Retrieval
    participant HS as HybridSearch
    participant VS as VectorSearch
    participant KS as KeywordSearch
    participant M as Memory
    
    A->>R: call(query, files)
    R->>R: 解析查询参数
    R->>M: 获取文档索引
    
    R->>HS: 执行混合搜索
    
    par 并行搜索
        HS->>VS: 向量相似度搜索
        VS->>VS: 计算查询向量
        VS->>VS: 相似度匹配
        VS-->>HS: 返回向量搜索结果
    and
        HS->>KS: BM25 关键词搜索
        KS->>KS: 分词和匹配
        KS-->>HS: 返回关键词搜索结果
    end
    
    HS->>HS: 结果融合和重排序
    HS->>HS: 应用过滤和截断
    HS-->>R: 返回最终检索结果
    
    R->>R: 格式化检索结果
    R-->>A: 返回结构化结果

flowchart TD
    Start[工具调用开始] --> Parse[参数解析]
    Parse --> Validate[参数验证]
    Validate -->|验证失败| Error[返回错误]
    Validate -->|验证成功| FileCheck{需要文件访问?}
    
    FileCheck -->|是| CopyFiles[复制文件到工作目录]
    FileCheck -->|否| Execute[执行工具逻辑]
    CopyFiles --> Execute
    
    Execute --> ToolType{工具类型}
    ToolType -->|代码解释器| CodeExec[代码执行]
    ToolType -->|网络搜索| WebSearch[网络搜索]
    ToolType -->|文档解析| DocParse[文档解析]
    ToolType -->|其他| OtherExec[其他执行]
    
    CodeExec --> KernelMgmt[内核管理]
    WebSearch --> HTTPReq[HTTP 请求]
    DocParse --> FileRead[文件读取]
    OtherExec --> ExtService[外部服务]
    
    KernelMgmt --> FormatResult[格式化结果]
    HTTPReq --> FormatResult
    FileRead --> FormatResult
    ExtService --> FormatResult
    
    FormatResult --> Return[返回结果]
    Error --> Return

class BaseTool(ABC):
    """工具基类，定义统一接口
    
    设计原则:
    - 统一的调用接口
    - 标准化的参数定义
    - OpenAI兼容的函数格式
    - 完善的错误处理
    """
    
    # 工具基本信息
    name: str = ''
    description: str = ''
    parameters: List[dict] = []
    
    # 工具特性标志
    file_access: bool = False  # 是否需要文件访问权限
    network_access: bool = False  # 是否需要网络访问权限
    
    def __init__(self, **kwargs):
        """初始化工具
        
        Args:
            **kwargs: 工具配置参数
        """
        self.cfg = kwargs
        
        # 验证工具名称
        if not self.name:
            raise ValueError(f'工具 {self.__class__.__name__} 必须设置 name 属性')
    
    @abstractmethod
    def call(self, params: str, **kwargs) -> Union[str, List[ContentItem]]:
        """
        工具调用的统一接口
        
        参数:
        - params: JSON格式的参数字符串
        - **kwargs: 额外的上下文参数
        
        返回:
        - Union[str, List[ContentItem]]: 工具执行结果
        
        实现要求:
        - 子类必须实现具体执行逻辑
        - 参数解析和验证
        - 异常处理和错误信息
        - 结果格式标准化
        """
        raise NotImplementedError
    
    @property
    def function(self) -> Dict:
        """
        生成OpenAI兼容的函数定义
        
        返回格式:
        {
            "name": "tool_name",
            "description": "工具描述",
            "parameters": {
                "type": "object",
                "properties": {...},
                "required": [...]
            }
        }
        """
        return {
            'name': self.name,
            'description': self.description,
            'parameters': {
                'type': 'object',
                'properties': {
                    param['name']: {
                        'type': param['type'],
                        'description': param['description'],
                        **{k: v for k, v in param.items() 
                           if k not in ['name', 'type', 'description', 'required']}
                    } for param in self.parameters
                },
                'required': [
                    param['name'] for param in self.parameters 
                    if param.get('required', False)
                ]
            }
        }
    
    def validate_params(self, params: dict) -> bool:
        """
        参数验证
        
        验证内容:
        - 必需参数检查
        - 参数类型验证
        - 参数值范围检查
        - 格式合规性验证
        """
        # 检查必需参数
        required_params = [p['name'] for p in self.parameters if p.get('required', False)]
        for param_name in required_params:
            if param_name not in params:
                raise ValueError(f"Missing required parameter: {param_name}")
        
        # 检查参数类型
        for param in self.parameters:
            param_name = param['name']
            if param_name in params:
                expected_type = param['type']
                actual_value = params[param_name]
                
                if not self._check_type(actual_value, expected_type):
                    raise ValueError(f"Parameter {param_name} should be {expected_type}, got {type(actual_value).__name__}")
        
        return True
    
    def _check_type(self, value, expected_type: str) -> bool:
        """类型检查"""
        type_mapping = {
            'string': str,
            'integer': int,
            'number': (int, float),
            'boolean': bool,
            'array': list,
            'object': dict
        }
        
        expected_python_type = type_mapping.get(expected_type)
        if expected_python_type:
            return isinstance(value, expected_python_type)
        
        return True
    
    def parse_params(self, params: str) -> dict:
        """
        解析参数字符串
        
        支持格式:
        - JSON字符串
        - URL编码格式
        - 简单键值对
        """
        import json5
        
        try:
            # 尝试JSON解析
            parsed_params = json5.loads(params)
            if isinstance(parsed_params, dict):
                return parsed_params
        except:
            pass
        
        try:
            # 尝试标准JSON解析
            parsed_params = json.loads(params)
            if isinstance(parsed_params, dict):
                return parsed_params
        except:
            pass
        
        # 简单键值对解析
        if '=' in params:
            result = {}
            for pair in params.split('&'):
                if '=' in pair:
                    key, value = pair.split('=', 1)
                    result[key.strip()] = value.strip()
            return result
        
        raise ValueError(f"Unable to parse parameters: {params}")
    
    def format_result(self, result: Any) -> Union[str, List[ContentItem]]:
        """
        格式化返回结果
        
        格式化策略:
        - 字符串直接返回
        - 字典/列表转JSON
        - 多媒体内容转ContentItem
        - 异常转错误信息
        """
        if isinstance(result, str):
            return result
        
        if isinstance(result, (dict, list)):
            import json
            return json.dumps(result, ensure_ascii=False, indent=2)
        
        if isinstance(result, Exception):
            return f"Error: {str(result)}"
        
        return str(result)

# 工具注册装饰器
def register_tool(name: str, allow_overwrite: bool = False):
    """
    工具注册装饰器
    
    功能:
    - 全局工具注册表管理
    - 重复注册检测
    - 名称冲突验证
    - 动态工具发现
    
    使用方式:
    @register_tool('tool_name')
    class MyTool(BaseTool):
        pass
    """
    def decorator(cls):
        # 类型检查
        if not issubclass(cls, BaseTool):
            raise ValueError(f"Class {cls.__name__} must inherit from BaseTool")
        
        # 重复注册检查
        if name in TOOL_REGISTRY:
            if allow_overwrite:
                logger.warning(f'Tool `{name}` already exists! Overwriting with class {cls}.')
            else:
                raise ValueError(f'Tool `{name}` already exists! Please ensure that the tool name is unique.')
        
        # 名称一致性检查
        if hasattr(cls, 'name') and cls.name and (cls.name != name):
            raise ValueError(f'{cls.__name__}.name="{cls.name}" conflicts with @register_tool(name="{name}").')
        
        # 注册工具
        cls.name = name
        TOOL_REGISTRY[name] = cls
        logger.info(f'Tool `{name}` registered successfully.')
        return cls
    
    return decorator

# 全局工具注册表
TOOL_REGISTRY: Dict[str, Type[BaseTool]] = {}

def get_tool(name: str, **kwargs) -> BaseTool:
    """
    工具工厂函数
    
    功能:
    - 根据名称获取工具实例
    - 支持参数传递
    - 工具初始化
    - 错误处理
    """
    if name not in TOOL_REGISTRY:
        raise ValueError(f"Tool '{name}' not found in registry. Available tools: {list(TOOL_REGISTRY.keys())}")
    
    tool_class = TOOL_REGISTRY[name]
    return tool_class(**kwargs)

def list_tools() -> List[str]:
    """列出所有已注册的工具"""
    return list(TOOL_REGISTRY.keys())

@register_tool('code_interpreter')
class CodeInterpreter(BaseTool):
    """代码解释器工具
    
    核心特性:
    - Jupyter内核管理
    - 安全代码执行
    - 多媒体输出支持
    - 会话状态保持
    """
    
    description = 'Python代码解释器，可以运行Python代码并返回结果，支持数据分析、可视化、计算等功能'
    parameters = [{
        'name': 'code',
        'type': 'string',
        'description': '要执行的Python代码',
        'required': True
    }]
    
    def __init__(self, timeout: int = 30, work_dir: str = None, **kwargs):
        """
        初始化代码解释器
        
        参数:
        - timeout: 代码执行超时时间（秒）
        - work_dir: 工作目录
        """
        super().__init__(**kwargs)
        self.timeout = timeout
        self.work_dir = work_dir or os.environ.get('M6_CODE_INTERPRETER_WORK_DIR', '/tmp/code_interpreter')
        
        # 确保工作目录存在
        os.makedirs(self.work_dir, exist_ok=True)
        
        # Jupyter内核管理
        self.kernel_manager = None
        self.kernel_client = None
        self.execution_count = 0
        
        # 初始化内核
        self._init_kernel()
    
    def call(self, params: str, **kwargs) -> str:
        """
        执行Python代码
        
        执行流程:
        1. 解析代码参数
        2. 代码安全检查
        3. 执行代码
        4. 收集执行结果
        5. 格式化输出
        """
        try:
            # 解析参数
            parsed_params = self.parse_params(params)
            self.validate_params(parsed_params)
            
            code = parsed_params['code']
            
            if not code.strip():
                return json.dumps({'error': 'Empty code provided'}, ensure_ascii=False)
            
            # 代码安全检查
            if not self._is_code_safe(code):
                return json.dumps({'error': 'Code contains potentially unsafe operations'}, ensure_ascii=False)
            
            # 执行代码
            result = self._execute_code_internal(code)
            
            # 格式化结果
            return self._format_execution_result(result)
            
        except Exception as e:
            error_result = {
                'error': str(e),
                'traceback': traceback.format_exc()
            }
            return json.dumps(error_result, ensure_ascii=False)

@abstractmethod
def call(self, params: Union[str, dict], **kwargs) -> Union[str, list, dict, List[ContentItem]]:
    """
    工具调用的统一抽象接口
    
    设计目的: 为所有工具提供标准化的调用方式和参数处理
    作用域: 所有工具实现的基础接口
    依赖前置条件: 工具已正确注册，参数格式符合要求
    后置条件: 返回工具执行结果，格式可为字符串、列表、字典或内容项列表
    复杂度: 取决于具体工具实现
    重要旁支: 子类必须实现此方法，定义具体的工具执行逻辑
    """
    raise NotImplementedError

def _verify_json_format_args(self, params: Union[str, dict], strict_json: bool = False) -> dict:
    """
    工具参数的格式验证和解析
    
    设计目的: 统一处理工具调用参数的格式验证和类型转换
    作用域: 所有工具的参数预处理
    依赖前置条件: 工具参数定义正确
    后置条件: 返回验证通过的参数字典
    复杂度: O(n) - n 为参数数量
    错误处理: JSON 解析失败或必需参数缺失时抛出 ValueError
    """
    # 参数格式统一化：字符串转字典
    if isinstance(params, str):
        try:
            if strict_json:
                params_json: dict = json.loads(params)  # 严格 JSON 解析
            else:
                params_json: dict = json_loads(params)  # 宽松 JSON 解析（支持 JSON5）
        except json.decoder.JSONDecodeError:
            raise ValueError('Parameters must be formatted as a valid JSON!')
    else:
        params_json: dict = params
    
    # 参数验证：根据工具定义验证参数
    if isinstance(self.parameters, list):
        # 列表格式的参数定义（旧格式）
        for param in self.parameters:
            if 'required' in param and param['required']:
                if param['name'] not in params_json:
                    raise ValueError('Parameters %s is required!' % param['name'])
    elif isinstance(self.parameters, dict):
        # 字典格式的参数定义（OpenAI 兼容格式）
        import jsonschema
        jsonschema.validate(instance=params_json, schema=self.parameters)
    else:
        raise ValueError
    
    return params_json

@register_tool('web_search')
class WebSearch(BaseTool):
    """网络搜索工具
    
    核心特性:
    - 多搜索引擎支持
    - 结果去重和排序
    - 内容提取和摘要
    - 搜索结果缓存
    """
    
    description = '网络搜索工具，可以搜索互联网上的信息并返回相关结果'
    parameters = [{
        'name': 'query',
        'type': 'string',
        'description': '搜索查询词',
        'required': True
    }, {
        'name': 'max_results',
        'type': 'integer',
        'description': '最大返回结果数量',
        'required': False,
        'default': 5
    }]
    
    network_access = True
    
    def call(self, params: str, **kwargs) -> str:
        """
        执行网络搜索
        
        执行流程:
        1. 解析搜索参数
        2. 检查缓存
        3. 执行搜索
        4. 结果处理和排序
        5. 格式化输出
        """
        try:
            # 解析参数
            parsed_params = self.parse_params(params)
            self.validate_params(parsed_params)
            
            query = parsed_params['query']
            max_results = parsed_params.get('max_results', 5)
            
            # 检查缓存
            cache_key = f"{query}:{max_results}"
            if cache_key in self.cache:
                cached_result, timestamp = self.cache[cache_key]
                if time.time() - timestamp < self.cache_ttl:
                    return cached_result
            
            # 执行搜索
            search_results = self._search(query, max_results)
            
            # 处理和格式化结果
            formatted_results = self._format_search_results(search_results)
            
            # 缓存结果
            self.cache[cache_key] = (formatted_results, time.time())
            
            return formatted_results
            
        except Exception as e:
            error_result = {
                'error': f'Search failed: {str(e)}',
                'query': parsed_params.get('query', ''),
                'results': []
            }
            return json.dumps(error_result, ensure_ascii=False)

# 全局工具注册表
TOOL_REGISTRY: Dict[str, Type[BaseTool]] = {}

class ToolManager:
    """工具管理器"""
    
    def __init__(self):
        self.tools: Dict[str, BaseTool] = {}
        self.tool_configs: Dict[str, dict] = {}
    
    def register_tool_class(self, name: str, tool_class: Type[BaseTool], config: dict = None):
        """注册工具类"""
        if not issubclass(tool_class, BaseTool):
            raise ValueError(f"Tool class must inherit from BaseTool")
        
        TOOL_REGISTRY[name] = tool_class
        if config:
            self.tool_configs[name] = config
        
        logger.info(f"Tool class '{name}' registered")
    
    def create_tool(self, name: str, **kwargs) -> BaseTool:
        """创建工具实例"""
        if name not in TOOL_REGISTRY:
            raise ValueError(f"Tool '{name}' not found")
        
        tool_class = TOOL_REGISTRY[name]
        
        # 合并配置
        config = self.tool_configs.get(name, {})
        config.update(kwargs)
        
        # 创建实例
        tool_instance = tool_class(**config)
        
        # 缓存实例
        self.tools[name] = tool_instance
        
        return tool_instance
    
    def get_tool(self, name: str) -> BaseTool:
        """获取工具实例"""
        if name in self.tools:
            return self.tools[name]
        
        return self.create_tool(name)
    
    def list_tools(self) -> List[str]:
        """列出所有可用工具"""
        return list(TOOL_REGISTRY.keys())
    
    def get_tool_info(self, name: str) -> dict:
        """获取工具信息"""
        if name not in TOOL_REGISTRY:
            raise ValueError(f"Tool '{name}' not found")
        
        tool_class = TOOL_REGISTRY[name]
        
        return {
            'name': name,
            'description': tool_class.description,
            'parameters': tool_class.parameters,
            'file_access': getattr(tool_class, 'file_access', False),
            'network_access': getattr(tool_class, 'network_access', False),
        }

# 全局工具管理器实例
tool_manager = ToolManager()

class ToolConfig:
    """工具配置管理"""
    
    def __init__(self, config_file: str = None):
        self.config_file = config_file
        self.configs = {}
        
        if config_file and os.path.exists(config_file):
            self.load_config()
    
    def load_config(self):
        """加载配置文件"""
        try:
            with open(self.config_file, 'r', encoding='utf-8') as f:
                self.configs = json.load(f)
        except Exception as e:
            logger.error(f"Failed to load tool config: {e}")
    
    def save_config(self):
        """保存配置文件"""
        if not self.config_file:
            return
        
        try:
            with open(self.config_file, 'w', encoding='utf-8') as f:
                json.dump(self.configs, f, ensure_ascii=False, indent=2)
        except Exception as e:
            logger.error(f"Failed to save tool config: {e}")
    
    def get_tool_config(self, tool_name: str) -> dict:
        """获取工具配置"""
        return self.configs.get(tool_name, {})
    
    def set_tool_config(self, tool_name: str, config: dict):
        """设置工具配置"""
        self.configs[tool_name] = config
        self.save_config()
    
    def update_tool_config(self, tool_name: str, updates: dict):
        """更新工具配置"""
        if tool_name not in self.configs:
            self.configs[tool_name] = {}
        
        self.configs[tool_name].update(updates)
        self.save_config()

# 默认工具配置
DEFAULT_TOOL_CONFIGS = {
    'code_interpreter': {
        'timeout': 30,
        'work_dir': '/tmp/code_interpreter'
    },
    'web_search': {
        'search_engine': 'bing',
        'max_results': 5
    },
    'retrieval': {
        'max_ref_token': 4000,
        'rag_searchers': ['hybrid_search']
    }
}

sequenceDiagram
    participant Agent as 智能体
    participant ToolManager as 工具管理器
    participant Tool as 工具实例
    participant Executor as 执行环境
    participant Storage as 存储系统
    
    Agent->>ToolManager: get_tool(tool_name)
    ToolManager->>ToolManager: 检查工具注册表
    ToolManager->>Tool: 创建工具实例
    Tool-->>ToolManager: 返回工具实例
    ToolManager-->>Agent: 返回工具实例
    
    Agent->>Tool: call(params)
    Tool->>Tool: 解析和验证参数
    Tool->>Executor: 执行具体操作
    
    alt 需要文件访问
        Executor->>Storage: 读取/写入文件
        Storage-->>Executor: 返回文件内容
    end
    
    alt 需要网络访问
        Executor->>Executor: 发起网络请求
    end
    
    Executor-->>Tool: 返回执行结果
    Tool->>Tool: 格式化结果
    Tool-->>Agent: 返回格式化结果

sequenceDiagram
    participant User as 用户
    participant CodeInterpreter as 代码解释器
    participant KernelManager as 内核管理器
    participant JupyterKernel as Jupyter内核
    participant FileSystem as 文件系统
    
    User->>CodeInterpreter: 执行代码请求
    CodeInterpreter->>CodeInterpreter: 代码安全检查
    CodeInterpreter->>KernelManager: 获取内核实例
    
    alt 内核未初始化
        KernelManager->>JupyterKernel: 启动新内核
        JupyterKernel-->>KernelManager: 内核就绪
    end
    
    KernelManager-->>CodeInterpreter: 返回内核客户端
    CodeInterpreter->>JupyterKernel: 发送执行请求
    
    loop 执行监控
        JupyterKernel-->>CodeInterpreter: 返回执行消息
        CodeInterpreter->>CodeInterpreter: 处理消息类型
        
        alt 生成图片
            CodeInterpreter->>FileSystem: 保存图片文件
            FileSystem-->>CodeInterpreter: 返回文件路径
        end
    end
    
    CodeInterpreter->>CodeInterpreter: 格式化执行结果
    CodeInterpreter-->>User: 返回执行结果

class ToolPool:
    """工具实例池"""
    
    def __init__(self, max_instances: int = 10):
        self.max_instances = max_instances
        self.pools: Dict[str, List[BaseTool]] = {}
        self.active_tools: Dict[str, List[BaseTool]] = {}
    
    def get_tool(self, tool_name: str) -> BaseTool:
        """从池中获取工具实例"""
        if tool_name not in self.pools:
            self.pools[tool_name] = []
            self.active_tools[tool_name] = []
        
        # 尝试从池中获取空闲实例
        if self.pools[tool_name]:
            tool = self.pools[tool_name].pop()
            self.active_tools[tool_name].append(tool)
            return tool
        
        # 检查是否达到最大实例数
        if len(self.active_tools[tool_name]) >= self.max_instances:
            # 等待或创建新实例
            pass
        
        # 创建新实例
        tool = tool_manager.create_tool(tool_name)
        self.active_tools[tool_name].append(tool)
        return tool
    
    def return_tool(self, tool: BaseTool):
        """归还工具实例到池中"""
        tool_name = tool.name
        
        if tool in self.active_tools.get(tool_name, []):
            self.active_tools[tool_name].remove(tool)
            
            # 重置工具状态
            if hasattr(tool, 'reset'):
                tool.reset()
            
            self.pools[tool_name].append(tool)

class PooledTool:
    """池化工具包装器"""
    
    def __init__(self, tool_name: str, pool: ToolPool):
        self.tool_name = tool_name
        self.pool = pool
        self.tool = None
    
    def __enter__(self):
        self.tool = self.pool.get_tool(self.tool_name)
        return self.tool
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.tool:
            self.pool.return_tool(self.tool)

# 使用示例
tool_pool = ToolPool()

with PooledTool('code_interpreter', tool_pool) as tool:
    result = tool.call('{"code": "print(\\"Hello World\\")"}')

class ToolCache:
    """工具结果缓存"""
    
    def __init__(self, max_size: int = 1000, ttl: int = 3600):
        self.max_size = max_size
        self.ttl = ttl
        self.cache: Dict[str, Tuple[Any, float]] = {}
        self.access_times: Dict[str, float] = {}
    
    def get_cache_key(self, tool_name: str, params: str, **kwargs) -> str:
        """生成缓存键"""
        import hashlib
        
        key_data = f"{tool_name}:{params}:{sorted(kwargs.items())}"
        return hashlib.md5(key_data.encode()).hexdigest()
    
    def get(self, key: str) -> Optional[Any]:
        """获取缓存结果"""
        if key not in self.cache:
            return None
        
        result, timestamp = self.cache[key]
        
        # 检查TTL
        if time.time() - timestamp > self.ttl:
            del self.cache[key]
            if key in self.access_times:
                del self.access_times[key]
            return None
        
        # 更新访问时间
        self.access_times[key] = time.time()
        return result
    
    def set(self, key: str, value: Any):
        """设置缓存结果"""
        # 检查缓存大小
        if len(self.cache) >= self.max_size:
            self._evict_oldest()
        
        self.cache[key] = (value, time.time())
        self.access_times[key] = time.time()
    
    def _evict_oldest(self):
        """淘汰最旧的缓存项"""
        if not self.access_times:
            return
        
        oldest_key = min(self.access_times.keys(), key=lambda k: self.access_times[k])
        del self.cache[oldest_key]
        del self.access_times[oldest_key]

class CachedTool(BaseTool):
    """带缓存的工具基类"""
    
    def __init__(self, cache: ToolCache = None, **kwargs):
        super().__init__(**kwargs)
        self.cache = cache or ToolCache()
    
    def call(self, params: str, **kwargs) -> Union[str, List[ContentItem]]:
        """带缓存的工具调用"""
        # 生成缓存键
        cache_key = self.cache.get_cache_key(self.name, params, **kwargs)
        
        # 检查缓存
        cached_result = self.cache.get(cache_key)
        if cached_result is not None:
            return cached_result
        
        # 执行工具调用
        result = self._call_impl(params, **kwargs)
        
        # 缓存结果
        self.cache.set(cache_key, result)
        
        return result
    
    @abstractmethod
    def _call_impl(self, params: str, **kwargs) -> Union[str, List[ContentItem]]:
        """实际的工具调用实现"""
        raise NotImplementedError

@register_tool('custom_tool')
class CustomTool(BaseTool):
    """自定义工具开发模板"""
    
    description = '自定义工具的描述'
    parameters = [{
        'name': 'param1',
        'type': 'string',
        'description': '参数1的描述',
        'required': True
    }, {
        'name': 'param2',
        'type': 'integer',
        'description': '参数2的描述',
        'required': False,
        'default': 10
    }]
    
    # 权限标志
    file_access = True
    network_access = False
    
    def __init__(self, custom_config: dict = None, **kwargs):
        super().__init__(**kwargs)
        self.custom_config = custom_config or {}
        
        # 初始化自定义资源
        self._init_resources()
    
    def _init_resources(self):
        """初始化自定义资源"""
        # 实现资源初始化逻辑
        pass
    
    def call(self, params: str, **kwargs) -> str:
        """
        工具调用实现
        
        开发步骤:
        1. 参数解析和验证
        2. 业务逻辑实现
        3. 结果格式化
        4. 错误处理
        """
        try:
            # 1. 参数解析和验证
            parsed_params = self.parse_params(params)
            self.validate_params(parsed_params)
            
            param1 = parsed_params['param1']
            param2 = parsed_params.get('param2', 10)
            
            # 2. 业务逻辑实现
            result = self._execute_business_logic(param1, param2)
            
            # 3. 结果格式化
            return self.format_result(result)
            
        except Exception as e:
            # 4. 错误处理
            return self._handle_error(e)
    
    def _execute_business_logic(self, param1: str, param2: int) -> dict:
        """执行业务逻辑"""
        # 实现具体的业务逻辑
        return {
            'param1': param1,
            'param2': param2,
            'result': f'Processed {param1} with {param2}'
        }
    
    def _handle_error(self, error: Exception) -> str:
        """错误处理"""
        error_info = {
            'error': str(error),
            'type': type(error).__name__,
            'tool': self.name
        }
        return json.dumps(error_info, ensure_ascii=False)
    
    def reset(self):
        """重置工具状态（用于池化）"""
        # 实现状态重置逻辑
        pass
    
    def __del__(self):
        """清理资源"""
        # 实现资源清理逻辑
        pass