flowchart TB
    subgraph "Tools 模块"
        TOOL[Tool 联合类型]
        
        subgraph "函数工具"
            FUNCTIONTOOL[FunctionTool]
            FUNCTOOLRESULT[FunctionToolResult]
            FUNCTOOLDECORATOR[@function_tool 装饰器]
        end
        
        subgraph "托管工具"
            FILESEARCH[FileSearchTool]
            WEBSEARCH[WebSearchTool]
            COMPUTER[ComputerTool]
            CODEINTERPRETER[CodeInterpreterTool]
            IMAGEGEN[ImageGenerationTool]
        end
        
        subgraph "本地工具"
            LOCALSHELL[LocalShellTool]
            LOCALSHELCMD[LocalShellCommandRequest]
        end
        
        subgraph "MCP 工具"
            HOSTEDMCP[HostedMCPTool]
            MCPAPPROVAL[MCPToolApprovalFunction]
            MCPREQUEST[MCPToolApprovalRequest]
        end
        
        subgraph "工具安全"
            TOOLINPUTGUARD[ToolInputGuardrail]
            TOOLOUTPUTGUARD[ToolOutputGuardrail]
            SAFETYCHECKDATA[ComputerToolSafetyCheckData]
        end
        
        subgraph "工具上下文"
            TOOLCONTEXT[ToolContext]
            RUNCONTEXT[RunContextWrapper]
        end
    end
    
    subgraph "支撑系统"
        FUNCTIONSCHEMA[function_schema 模式生成]
        STRICTSCHEMA[strict_schema 严格模式]
        COMPUTER_IMPL[computer 实现层]
        TRACING[tracing 追踪系统]
        GUARDRAILS[guardrails 安全框架]
    end
    
    TOOL --> FUNCTIONTOOL
    TOOL --> FILESEARCH
    TOOL --> WEBSEARCH
    TOOL --> COMPUTER
    TOOL --> CODEINTERPRETER
    TOOL --> IMAGEGEN
    TOOL --> LOCALSHELL
    TOOL --> HOSTEDMCP
    
    FUNCTIONTOOL --> FUNCTOOLRESULT
    FUNCTIONTOOL --> FUNCTOOLDECORATOR
    FUNCTIONTOOL --> TOOLINPUTGUARD
    FUNCTIONTOOL --> TOOLOUTPUTGUARD
    
    HOSTEDMCP --> MCPAPPROVAL
    HOSTEDMCP --> MCPREQUEST
    
    COMPUTER --> SAFETYCHECKDATA
    LOCALSHELL --> LOCALSHELCMD
    
    FUNCTIONTOOL --> TOOLCONTEXT
    FUNCTIONTOOL --> RUNCONTEXT
    
    FUNCTIONTOOL --> FUNCTIONSCHEMA
    FUNCTIONTOOL --> STRICTSCHEMA
    COMPUTER --> COMPUTER_IMPL
    FUNCTIONTOOL --> TRACING
    FUNCTIONTOOL --> GUARDRAILS
    
    style TOOL fill:#e1f5fe
    style FUNCTIONTOOL fill:#f3e5f5
    style HOSTEDMCP fill:#e8f5e8
    style TOOLCONTEXT fill:#fff3e0

def function_tool(
    func: ToolFunction | None = None,
    *,
    name_override: str | None = None,
    description_override: str | None = None,
    strict_json_schema: bool = True,
    is_enabled: bool | Callable = True,
    **kwargs
) -> Callable:
    """函数工具装饰器的核心实现"""
    
    def decorator(target_func: ToolFunction) -> FunctionTool:
        # 1) 分析函数签名，确定参数类型
        sig = inspect.signature(target_func)
        params = list(sig.parameters.values())
        
        # 2) 检测上下文参数类型
        context_param = None
        func_params = []
        
        for param in params:
            if param.annotation in [RunContextWrapper, ToolContext]:
                context_param = param
            else:
                func_params.append(param)
        
        # 3) 生成 JSON Schema
        schema = function_schema(
            target_func,
            name_override=name_override,
            description_override=description_override,
            docstring_style=DocstringStyle.GOOGLE,
            strict_json_schema=strict_json_schema
        )
        
        # 4) 创建工具调用包装器
        async def tool_invoker(tool_context: ToolContext, args_json: str) -> Any:
            try:
                # 解析 JSON 参数
                args = json.loads(args_json)
                
                # 构建函数调用参数
                call_args = []
                if context_param:
                    if context_param.annotation == ToolContext:
                        call_args.append(tool_context)
                    else:  # RunContextWrapper
                        call_args.append(tool_context.run_context)
                
                # 添加业务参数
                for param in func_params:
                    if param.name in args:
                        call_args.append(args[param.name])
                    elif param.default is not param.empty:
                        call_args.append(param.default)
                    else:
                        raise ValueError(f"Missing required parameter: {param.name}")
                
                # 执行原函数
                if inspect.iscoroutinefunction(target_func):
                    result = await target_func(*call_args)
                else:
                    result = target_func(*call_args)
                
                return result
                
            except Exception as e:
                # 错误处理和日志记录
                logger.error(f"Tool execution failed: {e}")
                return f"Tool execution error: {str(e)}"
        
        # 5) 创建 FunctionTool 实例
        return FunctionTool(
            name=schema["name"],
            description=schema["description"],
            params_json_schema=schema["parameters"],
            on_invoke_tool=tool_invoker,
            strict_json_schema=strict_json_schema,
            is_enabled=is_enabled
        )
    
    # 支持直接调用和装饰器两种用法
    if func is not None:
        return decorator(func)
    return decorator

async def execute_tool_with_guardrails(
    tool: FunctionTool,
    context: ToolContext,
    args_json: str
) -> FunctionToolResult:
    """带安全检查的工具执行流程"""
    
    # 1) 输入安全检查
    if tool.tool_input_guardrails:
        input_data = ToolInputGuardrailData(
            tool_name=tool.name,
            tool_input=args_json,
            context=context.run_context.context
        )
        
        for guardrail in tool.tool_input_guardrails:
            try:
                with guardrail_span(guardrail.__class__.__name__):
                    result = await _coro.ensure_awaitable(
                        guardrail(input_data, context.run_context, context.agent)
                    )
                    
                    if not result.passed:
                        raise ToolInputGuardrailTripwireTriggered(
                            guardrail_name=guardrail.__class__.__name__,
                            failure_reason=result.failure_reason,
                            tool_name=tool.name
                        )
            except Exception as e:
                logger.error(f"Input guardrail {guardrail.__class__.__name__} failed: {e}")
                raise
    
    # 2) 执行工具
    try:
        with function_span(tool.name) as span:
            raw_output = await tool.on_invoke_tool(context, args_json)
            span.add_input(args_json)
            span.add_output(str(raw_output))
    except Exception as e:
        logger.error(f"Tool {tool.name} execution failed: {e}")
        raise
    
    # 3) 输出安全检查
    if tool.tool_output_guardrails:
        output_data = ToolOutputGuardrailData(
            tool_name=tool.name,
            tool_input=args_json,
            tool_output=str(raw_output),
            context=context.run_context.context
        )
        
        for guardrail in tool.tool_output_guardrails:
            try:
                with guardrail_span(guardrail.__class__.__name__):
                    result = await _coro.ensure_awaitable(
                        guardrail(output_data, context.run_context, context.agent)
                    )
                    
                    if not result.passed:
                        raise ToolOutputGuardrailTripwireTriggered(
                            guardrail_name=guardrail.__class__.__name__,
                            failure_reason=result.failure_reason,
                            tool_name=tool.name
                        )
            except Exception as e:
                logger.error(f"Output guardrail {guardrail.__class__.__name__} failed: {e}")
                raise
    
    # 4) 创建结果对象
    run_item = ToolCallOutputItem(
        tool_name=tool.name,
        input=args_json,
        output=str(raw_output)
    )
    
    return FunctionToolResult(
        tool=tool,
        output=raw_output,
        run_item=run_item
    )

class MCPUtil:
    """MCP 工具集成工具类"""
    
    @staticmethod
    async def get_all_function_tools(
        servers: list[MCPServer],
        convert_schemas_to_strict: bool,
        run_context: RunContextWrapper,
        agent: AgentBase
    ) -> list[Tool]:
        """从 MCP 服务器获取所有工具"""
        all_tools = []
        
        for server in servers:
            try:
                # 1) 连接服务器并获取工具列表
                await server.connect()
                mcp_tools = await server.list_tools(run_context, agent)
                
                for mcp_tool in mcp_tools:
                    # 2) 转换 MCP 工具为 FunctionTool
                    function_tool = await MCPUtil._convert_mcp_tool_to_function_tool(
                        server, mcp_tool, convert_schemas_to_strict
                    )
                    all_tools.append(function_tool)
                    
            except Exception as e:
                logger.warning(f"Failed to get tools from MCP server {server.name}: {e}")
                continue  # 跳过失败的服务器，不影响其他服务器
        
        return all_tools
    
    @staticmethod
    async def _convert_mcp_tool_to_function_tool(
        server: MCPServer,
        mcp_tool: MCPTool,
        convert_to_strict: bool
    ) -> FunctionTool:
        """将 MCP 工具转换为 FunctionTool"""
        
        # 1) 处理 JSON Schema
        schema = mcp_tool.inputSchema
        if convert_to_strict and schema:
            try:
                schema = ensure_strict_json_schema(schema)
            except Exception as e:
                logger.warning(f"Failed to convert MCP tool {mcp_tool.name} to strict schema: {e}")
                # 降级使用原始 schema
        
        # 2) 创建工具调用包装器
        async def mcp_tool_invoker(context: ToolContext, args_json: str) -> str:
            try:
                # 解析参数
                args = json.loads(args_json) if args_json else {}
                
                # 调用 MCP 服务器
                with mcp_tools_span(server.name, mcp_tool.name) as span:
                    result = await server.call_tool(mcp_tool.name, args)
                    span.add_input(args_json)
                    span.add_output(result.content)
                
                # 处理结构化内容
                if server.use_structured_content and result.structured_content:
                    # 使用结构化内容
                    return json.dumps(result.structured_content)
                else:
                    # 使用文本内容
                    return str(result.content)
                
            except Exception as e:
                logger.error(f"MCP tool {mcp_tool.name} execution failed: {e}")
                return f"MCP tool execution error: {str(e)}"
        
        # 3) 创建 FunctionTool 实例
        return FunctionTool(
            name=mcp_tool.name,
            description=mcp_tool.description or f"MCP tool: {mcp_tool.name}",
            params_json_schema=schema or {},
            on_invoke_tool=mcp_tool_invoker,
            strict_json_schema=convert_to_strict
        )

@function_tool
def calculate_fibonacci(n: int) -> int:
    """计算斐波那契数列的第 n 项
    
    Args:
        n: 要计算的项数，必须是正整数
        
    Returns:
        斐波那契数列的第 n 项
    """
    if n <= 0:
        return 0
    elif n == 1:
        return 1
    else:
        a, b = 0, 1
        for _ in range(2, n + 1):
            a, b = b, a + b
        return b

FileSearchTool(
    vector_store_ids=["vs_123", "vs_456"],
    max_num_results=5,
    ranking_options=RankingOptions(score_threshold=0.8),
    filters=Filters(filename_contains="report")
)

WebSearchTool(
    user_location=UserLocation(country="US", city="San Francisco"),
    search_context_size="high",
    filters=WebSearchToolFilters(domain_includes=["wikipedia.org"])
)

ComputerTool(
    display_width_px=1920,
    display_height_px=1080,
    computer_impl=AsyncComputer()  # 或自定义实现
)

LocalShellTool(
    # 默认配置，支持基本的 shell 命令执行
)

# 使用示例
shell_tool = LocalShellTool()
# 代理可以调用："执行命令 'ls -la' 查看当前目录文件"

HostedMCPTool(
    tool_config=Mcp(
        server_url="https://mcp-server.example.com",
        api_key="your-api-key"
    ),
    on_approval_request=approval_handler  # 可选的审批处理函数
)

from agents import function_tool
from typing import Optional, List

@function_tool
async def search_products(
    query: str,
    category: Optional[str] = None,
    max_results: int = 10,
    sort_by: str = "relevance"
) -> List[dict]:
    """搜索产品信息
    
    Args:
        query: 搜索关键词，必填
        category: 产品分类，可选
        max_results: 最大结果数量，默认10条
        sort_by: 排序方式，可选值：relevance, price, rating
        
    Returns:
        包含产品信息的字典列表，每个字典包含 id, name, price, rating 字段
    """
    # 实际搜索逻辑
    results = await product_search_api(
        query=query,
        category=category,
        limit=max_results,
        sort=sort_by
    )
    
    return [
        {
            "id": item.id,
            "name": item.name,
            "price": item.price,
            "rating": item.rating
        }
        for item in results
    ]

# 最佳实践要点：
# 1. 详细的文档字符串，说明参数和返回值
# 2. 合理的默认参数，提高易用性
# 3. 类型注解完整，支持自动 Schema 生成
# 4. 返回结构化数据，便于 LLM 理解和处理

from agents import function_tool, tool_input_guardrail, tool_output_guardrail

@tool_input_guardrail
def validate_file_path(data, context, agent):
    """验证文件路径的安全性"""
    import os
    args = json.loads(data.tool_input)
    file_path = args.get("file_path", "")
    
    # 检查路径遍历攻击
    if ".." in file_path or file_path.startswith("/"):
        return ToolInputGuardrailResult(
            passed=False,
            failure_reason="不允许访问上级目录或根目录"
        )
    
    # 检查文件扩展名
    allowed_extensions = [".txt", ".json", ".csv", ".md"]
    if not any(file_path.endswith(ext) for ext in allowed_extensions):
        return ToolInputGuardrailResult(
            passed=False,
            failure_reason="只允许访问文本文件"
        )
    
    return ToolInputGuardrailResult(passed=True)

@tool_output_guardrail
def filter_sensitive_content(data, context, agent):
    """过滤输出中的敏感信息"""
    import re
    
    sensitive_patterns = [
        r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b',  # 信用卡号
        r'\b\d{3}-\d{2}-\d{4}\b',  # 社会安全号
    ]
    
    filtered_output = data.tool_output
    for pattern in sensitive_patterns:
        filtered_output = re.sub(pattern, "[REDACTED]", filtered_output)
    
    if filtered_output != data.tool_output:
        logger.info("Sensitive content filtered from tool output")
    
    return ToolOutputGuardrailResult(
        passed=True,
        modified_output=filtered_output
    )

@function_tool(
    tool_input_guardrails=[validate_file_path],
    tool_output_guardrails=[filter_sensitive_content]
)
def read_file_safe(file_path: str) -> str:
    """安全地读取文件内容"""
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            return f.read()
    except Exception as e:
        return f"读取文件失败：{str(e)}"

from agents import Agent, function_tool

class DynamicToolManager:
    """动态工具管理器"""
    
    def __init__(self):
        self.user_permissions = {}
        self.tool_registry = {}
    
    def register_tool(self, tool_name: str, tool_func):
        """注册工具"""
        self.tool_registry[tool_name] = tool_func
    
    def get_enabled_tools(self, user_id: str) -> List[Tool]:
        """获取用户可用的工具列表"""
        user_perms = self.user_permissions.get(user_id, set())
        enabled_tools = []
        
        for tool_name, tool_func in self.tool_registry.items():
            if tool_name in user_perms:
                enabled_tools.append(tool_func)
        
        return enabled_tools

# 使用示例
tool_manager = DynamicToolManager()

@function_tool(
    is_enabled=lambda context, agent: (
        context.context.get("user_role") == "admin"
    )
)
def delete_user(user_id: str) -> str:
    """删除用户（仅管理员可用）"""
    # 删除逻辑
    return f"用户 {user_id} 已删除"

@function_tool(
    is_enabled=lambda context, agent: (
        context.context.get("subscription_level", "free") in ["premium", "enterprise"]
    )
)
def advanced_analytics(query: str) -> str:
    """高级分析功能（仅付费用户可用）"""
    # 分析逻辑
    return f"分析结果：{query}"

# 创建具有条件工具的代理
agent = Agent(
    name="ConditionalAgent",
    tools=[delete_user, advanced_analytics],
    instructions="根据用户权限提供相应的功能"
)

@function_tool
async def comprehensive_research(topic: str) -> str:
    """综合研究：结合多个工具进行深度分析"""
    
    # 1. 网络搜索获取基础信息
    web_results = await web_search(f"{topic} overview recent developments")
    
    # 2. 文档检索获取相关资料
    doc_results = await document_search(topic, max_results=5)
    
    # 3. 数据分析
    analysis_data = await analyze_trends(topic)
    
    # 4. 生成综合报告
    report = f"""
    # {topic} 综合研究报告
    
    ## 网络搜索结果摘要
    {web_results[:500]}...
    
    ## 相关文档发现

    - 找到 {len(doc_results)} 篇相关文档
    - 关键主题：{extract_key_themes(doc_results)}
    
    ## 趋势分析
    {analysis_data}
    
    ## 结论与建议
    {generate_recommendations(web_results, doc_results, analysis_data)}
    """
    
    return report

# 工具组合的优势：
# 1. 多数据源整合，提供全面视角
# 2. 自动化工作流，提高效率
# 3. 结构化输出，便于后续处理

Tool (联合类型)
    ├── FunctionTool (函数工具)
    │   └── function_tool() 装饰器/工厂函数
    ├── FileSearchTool (文件搜索工具)
    ├── WebSearchTool (网页搜索工具)
    ├── ComputerTool (计算机控制工具)
    ├── HostedMCPTool (托管MCP工具)
    ├── LocalShellTool (本地Shell工具)
    ├── CodeInterpreterTool (代码解释器工具)
    └── ImageGenerationTool (图像生成工具)

# 装饰器用法（无参数）
@function_tool
def my_tool(param1: str, param2: int) -> str:
    ...

# 装饰器用法（带参数）
@function_tool(
    name_override="custom_name",
    description_override="Custom description",
    strict_mode=True,
    is_enabled=True
)
def my_tool(param1: str, param2: int) -> str:
    ...

# 工厂函数用法
tool = function_tool(my_function, name_override="custom_name")

FunctionTool  # 可直接传递给 Agent(tools=[...])

from agents import Agent, function_tool, RunContextWrapper

# 基础用法：简单函数
@function_tool
def get_current_time() -> str:
    """获取当前时间"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

# 带参数的函数
@function_tool
def calculate(expression: str) -> str:
    """
    计算数学表达式的结果
    
    Args:
        expression: 要计算的数学表达式，例如 "2 + 2"
    
    Returns:
        计算结果的字符串表示
    """
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算错误: {e}"

# 带上下文的函数
@function_tool
def query_database(
    sql: str,
    context: RunContextWrapper
) -> str:
    """
    执行数据库查询
    
    Args:
        sql: SQL查询语句
        context: 运行上下文，包含数据库连接
    
    Returns:
        查询结果
    """
    db = context.context.database  # 访问用户提供的数据库连接
    results = db.execute(sql)
    return str(results)

# 自定义配置
@function_tool(
    name_override="search_product",
    description_override="在产品数据库中搜索商品",
    strict_mode=True
)
def product_search(keyword: str, limit: int = 10) -> str:
    """搜索商品"""
    # 实际搜索逻辑
    return f"找到 {limit} 个包含'{keyword}'的商品"

# 动态启用/禁用
def is_premium_user(context: RunContextWrapper, agent) -> bool:
    """只有高级用户可以使用此工具"""
    return context.context.get("subscription") == "premium"

@function_tool(is_enabled=is_premium_user)
def premium_feature(param: str) -> str:
    """高级功能，仅限高级用户"""
    return f"执行高级功能: {param}"

# 自定义错误处理
def custom_error_handler(context: RunContextWrapper, error: Exception) -> str:
    """自定义错误消息"""
    return f"操作失败，请稍后重试。技术详情已记录（错误ID: {id(error)}）"

@function_tool(failure_error_function=custom_error_handler)
def risky_operation(data: str) -> str:
    """可能失败的操作"""
    # 可能抛出异常的操作
    return process_data(data)

# 创建代理并使用工具
agent = Agent(
    name="ToolUser",
    instructions="你可以使用多种工具来完成任务。",
    tools=[
        get_current_time,
        calculate,
        query_database,
        product_search,
        premium_feature
    ]
)

# 基本类型
def tool(
    text: str,           # 字符串
    number: int,         # 整数
    decimal: float,      # 浮点数
    flag: bool,          # 布尔值
) -> str:
    ...

# 复杂类型
def tool(
    items: list[str],    # 列表
    config: dict,        # 字典
    optional: str | None # 可选参数
) -> str:
    ...

# 带默认值
def tool(
    required: str,
    optional: int = 10,
    flag: bool = False
) -> str:
    ...

# 带上下文（自动注入，不传递给LLM）
def tool(
    param: str,
    context: RunContextWrapper  # 自动注入
) -> str:
    ...

def tool(
    param: str,
    context: ToolContext  # 包含更多信息的上下文
) -> str:
    ...

@function_tool
def may_fail(param: str) -> str:
    """可能失败的工具"""
    try:
        # 执行操作
        result = risky_operation(param)
        return str(result)
    except ValueError as e:
        # 返回错误消息（模型可以看到）
        return f"参数错误: {e}"
    except Exception as e:
        # 抛出异常会中断执行
        raise RuntimeError(f"严重错误: {e}")

@dataclass
class FunctionTool:
    name: str                    # 工具名称
    description: str             # 工具描述
    params_json_schema: dict     # 参数JSON Schema
    on_invoke_tool: Callable     # 调用函数
    strict_json_schema: bool = True  # 严格模式
    is_enabled: bool | Callable = True  # 是否启用
    tool_input_guardrails: list | None = None  # 输入防护
    tool_output_guardrails: list | None = None  # 输出防护

from agents.tool import FunctionTool
from agents.function_schema import function_schema

def my_function(param1: str, param2: int) -> str:
    return f"{param1}: {param2}"

# 手动创建FunctionTool
tool = FunctionTool(
    name="my_tool",
    description="这是一个示例工具",
    params_json_schema=function_schema(my_function),
    on_invoke_tool=lambda ctx, args_json: my_function(**json.loads(args_json)),
    strict_json_schema=True,
    is_enabled=True
)

@dataclass
class FileSearchTool:
    def __init__(
        self,
        vector_store_ids: list[str],
        max_num_results: int | None = None,
        include_search_results: bool = False,
        ranking_options: RankingOptions | None = None,
        filters: Filters | None = None
    )

from agents import Agent, FileSearchTool

# 创建文件搜索工具
file_search = FileSearchTool(
    vector_store_ids=["vs_abc123", "vs_def456"],
    max_num_results=10,
    include_search_results=True
)

# 创建带文件搜索的代理
agent = Agent(
    name="DocumentAssistant",
    instructions="你可以搜索文档库来回答问题。",
    tools=[file_search]
)

# 代理会自动使用file_search工具查找相关文档
result = await Runner.run(
    agent,
    "请查找关于产品定价的文档"
)

@dataclass
class WebSearchTool:
    def __init__(
        self,
        user_location: UserLocation | None = None,
        filters: WebSearchToolFilters | None = None,
        search_context_size: Literal["low", "medium", "high"] = "medium"
    )

from agents import Agent, WebSearchTool
from openai.types.responses.web_search_tool_param import UserLocation

# 创建网页搜索工具
web_search = WebSearchTool(
    user_location=UserLocation(
        city="Beijing",
        country="CN"
    ),
    search_context_size="high"
)

# 创建搜索代理
agent = Agent(
    name="ResearchAssistant",
    instructions="你可以搜索网页获取最新信息。",
    tools=[web_search]
)

# 代理会自动搜索网页
result = await Runner.run(
    agent,
    "最新的AI技术发展趋势是什么？"
)

@dataclass
class CodeInterpreterTool:
    def __init__(
        self,
        tool_config: CodeInterpreter
    )

from agents import Agent, CodeInterpreterTool
from openai.types.responses.tool_param import CodeInterpreter

# 创建代码解释器工具
code_interpreter = CodeInterpreterTool(
    tool_config=CodeInterpreter(
        container="default"  # 沙箱容器配置
    )
)

# 创建数据分析代理
agent = Agent(
    name="DataAnalyst",
    instructions="你可以编写和执行Python代码进行数据分析。",
    tools=[code_interpreter]
)

# 代理会自动编写和执行代码
result = await Runner.run(
    agent,
    "请分析这组数据的统计特征：[1, 2, 3, 4, 5, 10, 20, 30]"
)

@dataclass
class ImageGenerationTool:
    def __init__(
        self,
        tool_config: ImageGeneration
    )

from agents import Agent, ImageGenerationTool
from openai.types.responses.tool_param import ImageGeneration

# 创建图像生成工具
image_gen = ImageGenerationTool(
    tool_config=ImageGeneration()
)

# 创建创意代理
agent = Agent(
    name="CreativeAssistant",
    instructions="你可以根据描述生成图像。",
    tools=[image_gen]
)

# 代理会自动生成图像
result = await Runner.run(
    agent,
    "请创建一张未来城市的概念图"
)

@dataclass
class ComputerTool:
    def __init__(
        self,
        computer: Computer | AsyncComputer,
        on_safety_check: Callable[[ComputerToolSafetyCheckData], bool] | None = None
    )

from agents import Agent, ComputerTool
from agents.computer import Computer

class MyComputer(Computer):
    """自定义计算机实现"""
    
    @property
    def display_width_px(self) -> int:
        return 1920
    
    @property
    def display_height_px(self) -> int:
        return 1080
    
    async def screenshot(self) -> bytes:
        """截取屏幕"""
        # 实现截图逻辑
        return screenshot_bytes
    
    async def mouse_move(self, x: int, y: int):
        """移动鼠标"""
        # 实现鼠标移动
        pass
    
    async def mouse_click(self, button: str):
        """点击鼠标"""
        # 实现鼠标点击
        pass
    
    async def keyboard_type(self, text: str):
        """输入文本"""
        # 实现键盘输入
        pass

# 安全检查回调
def safety_check(data: ComputerToolSafetyCheckData) -> bool:
    """检查计算机操作是否安全"""
    action = data.tool_call.action
    
    # 阻止危险操作
    if action == "delete_file":
        return False
    
    # 允许其他操作
    return True

# 创建计算机工具
computer_tool = ComputerTool(
    computer=MyComputer(),
    on_safety_check=safety_check
)

# 创建自动化代理
agent = Agent(
    name="AutomationAgent",
    instructions="你可以控制计算机完成任务。",
    tools=[computer_tool]
)

@dataclass
class HostedMCPTool:
    def __init__(
        self,
        tool_config: Mcp,
        on_approval_request: MCPToolApprovalFunction | None = None
    )

from agents import Agent, HostedMCPTool
from openai.types.responses.tool_param import Mcp

# 批准函数
async def approve_mcp_tool(
    request: MCPToolApprovalRequest
) -> MCPToolApprovalFunctionResult:
    """审批MCP工具调用"""
    
    tool_name = request.data.tool_name
    
    # 自动批准某些工具
    if tool_name in ["read_file", "list_directory"]:
        return {"approve": True}
    
    # 拒绝敏感工具
    if tool_name == "delete_file":
        return {
            "approve": False,
            "reason": "删除文件需要人工批准"
        }
    
    # 默认批准
    return {"approve": True}

# 创建托管MCP工具
mcp_tool = HostedMCPTool(
    tool_config=Mcp(
        server_url="https://mcp.example.com",
        api_key="your-api-key"
    ),
    on_approval_request=approve_mcp_tool
)

# 创建集成代理
agent = Agent(
    name="MCPAgent",
    instructions="你可以使用MCP服务器的工具。",
    tools=[mcp_tool]
)

@dataclass
class LocalShellTool:
    def __init__(
        self,
        executor: LocalShellExecutor
    )

from agents import Agent, LocalShellTool
import subprocess

# 命令执行器
async def shell_executor(
    request: LocalShellCommandRequest
) -> str:
    """执行Shell命令"""
    
    command = request.data.command
    
    # 安全检查
    if any(danger in command for danger in ["rm -rf", "format", "del"]):
        return "错误：命令包含危险操作"
    
    try:
        # 执行命令
        result = subprocess.run(
            command,
            shell=True,
            capture_output=True,
            text=True,
            timeout=30
        )
        
        return result.stdout or result.stderr
    
    except subprocess.TimeoutExpired:
        return "错误：命令执行超时"
    except Exception as e:
        return f"错误：{e}"

# 创建本地Shell工具
shell_tool = LocalShellTool(executor=shell_executor)

# 创建系统管理代理
agent = Agent(
    name="SysAdminAgent",
    instructions="你可以执行系统命令完成任务。",
    tools=[shell_tool]
)

from agents.tool_guardrails import ToolInputGuardrail, ToolInputGuardrailResult

class ParameterValidationGuardrail(ToolInputGuardrail):
    """参数验证防护"""
    
    async def run(
        self,
        tool_name: str,
        arguments: dict,
        context: Any
    ) -> ToolInputGuardrailResult:
        """检查工具参数"""
        
        # 检查参数范围
        if "amount" in arguments:
            amount = arguments["amount"]
            if amount > 1000:
                return ToolInputGuardrailResult(
                    output=ToolGuardrailFunctionOutput(
                        behavior=RejectContentBehavior(
                            reject_message="金额超过限制"
                        )
                    )
                )
        
        # 参数合法
        return ToolInputGuardrailResult(
            output=ToolGuardrailFunctionOutput(
                behavior=AllowBehavior()
            )
        )

# 应用到工具
@function_tool
def transfer_money(amount: float, to_account: str) -> str:
    """转账"""
    return f"转账 {amount} 到 {to_account}"

# 手动添加防护
transfer_money.tool_input_guardrails = [ParameterValidationGuardrail()]

from agents.tool_guardrails import ToolOutputGuardrail, ToolOutputGuardrailResult

class SensitiveDataGuardrail(ToolOutputGuardrail):
    """敏感数据防护"""
    
    async def run(
        self,
        tool_name: str,
        output: str,
        context: Any
    ) -> ToolOutputGuardrailResult:
        """检查工具输出"""
        
        # 检查敏感信息
        if contains_credit_card(output):
            return ToolOutputGuardrailResult(
                output=ToolGuardrailFunctionOutput(
                    behavior=RejectContentBehavior(
                        reject_message="输出包含信用卡信息"
                    )
                )
            )
        
        # 输出安全
        return ToolOutputGuardrailResult(
            output=ToolGuardrailFunctionOutput(
                behavior=AllowBehavior()
            )
        )

# 定义工具集
basic_tools = [
    get_current_time,
    calculate
]

advanced_tools = [
    query_database,
    send_email
]

# 创建不同能力的代理
basic_agent = Agent(
    name="BasicAgent",
    tools=basic_tools
)

advanced_agent = Agent(
    name="AdvancedAgent",
    tools=basic_tools + advanced_tools
)

def tool_availability(context: RunContextWrapper, agent) -> bool:
    """根据上下文动态控制工具可用性"""
    
    # 根据用户权限
    user_role = context.context.get("role")
    if user_role == "admin":
        return True
    
    # 根据时间
    hour = datetime.now().hour
    if 9 <= hour <= 17:  # 工作时间
        return True
    
    return False

@function_tool(is_enabled=tool_availability)
def admin_operation(action: str) -> str:
    """管理员操作"""
    return f"执行管理操作: {action}"

Tool (联合类型)
    ├── FunctionTool (函数工具)
    │   ├── 参数 JSON Schema
    │   ├── 执行函数
    │   └── 防护配置
    ├── 托管工具
    │   ├── FileSearchTool
    │   ├── WebSearchTool
    │   ├── CodeInterpreterTool
    │   └── ImageGenerationTool
    └── 高级工具
        ├── ComputerTool
        ├── HostedMCPTool
        └── LocalShellTool

工具执行上下文
    ├── ToolContext
    └── RunContextWrapper

工具结果
    └── FunctionToolResult

classDiagram
    class Tool {
        <<Union Type>>
        FunctionTool | FileSearchTool | WebSearchTool | ComputerTool | HostedMCPTool | LocalShellTool | CodeInterpreterTool | ImageGenerationTool
    }
    
    class FunctionTool {
        +str name
        +str description
        +dict params_json_schema
        +Callable on_invoke_tool
        +bool strict_json_schema
        +bool | Callable is_enabled
        +list | None tool_input_guardrails
        +list | None tool_output_guardrails
        
        +__post_init__()
    }
    
    class FileSearchTool {
        +list[str] vector_store_ids
        +int | None max_num_results
        +bool include_search_results
        +RankingOptions | None ranking_options
        +Filters | None filters
        
        +name() str
    }
    
    class WebSearchTool {
        +UserLocation | None user_location
        +WebSearchToolFilters | None filters
        +Literal["low","medium","high"] search_context_size
        
        +name() str
    }
    
    class ComputerTool {
        +Computer | AsyncComputer computer
        +Callable | None on_safety_check
        
        +name() str
    }
    
    class HostedMCPTool {
        +Mcp tool_config
        +MCPToolApprovalFunction | None on_approval_request
        
        +name() str
    }
    
    class LocalShellTool {
        +LocalShellExecutor executor
        
        +name() str
    }
    
    class CodeInterpreterTool {
        +CodeInterpreter tool_config
        
        +name() str
    }
    
    class ImageGenerationTool {
        +ImageGeneration tool_config
        
        +name() str
    }
    
    Tool --> FunctionTool
    Tool --> FileSearchTool
    Tool --> WebSearchTool
    Tool --> ComputerTool
    Tool --> HostedMCPTool
    Tool --> LocalShellTool
    Tool --> CodeInterpreterTool
    Tool --> ImageGenerationTool
    
    note for Tool "工具联合类型<br/>支持多种工具实现<br/>统一的工具接口"
    note for FunctionTool "Python函数包装<br/>最灵活的工具类型<br/>支持防护和动态启用"
    note for FileSearchTool "向量存储搜索<br/>OpenAI托管服务<br/>语义搜索文档"
    note for ComputerTool "计算机控制<br/>鼠标键盘操作<br/>自动化任务"

classDiagram
    class FunctionTool {
        +str name
        +str description
        +dict params_json_schema
        +Callable on_invoke_tool
        +bool strict_json_schema
        +bool | Callable is_enabled
        +list~ToolInputGuardrail~ | None tool_input_guardrails
        +list~ToolOutputGuardrail~ | None tool_output_guardrails
    }
    
    class ToolFunction {
        <<Union Type>>
        ToolFunctionWithoutContext | ToolFunctionWithContext | ToolFunctionWithToolContext
    }
    
    class ToolFunctionWithoutContext {
        <<Callable>>
        (ToolParams) -> Any
    }
    
    class ToolFunctionWithContext {
        <<Callable>>
        (RunContextWrapper, ToolParams) -> Any
    }
    
    class ToolFunctionWithToolContext {
        <<Callable>>
        (ToolContext, ToolParams) -> Any
    }
    
    class ToolInputGuardrail {
        <<Abstract>>
        +run(tool_name, arguments, context)* ToolInputGuardrailResult
    }
    
    class ToolOutputGuardrail {
        <<Abstract>>
        +run(tool_name, output, context)* ToolOutputGuardrailResult
    }
    
    class FunctionToolResult {
        +FunctionTool tool
        +Any output
        +RunItem run_item
    }
    
    FunctionTool --> ToolFunction : wraps
    ToolFunction --> ToolFunctionWithoutContext
    ToolFunction --> ToolFunctionWithContext
    ToolFunction --> ToolFunctionWithToolContext
    
    FunctionTool --> ToolInputGuardrail : may have
    FunctionTool --> ToolOutputGuardrail : may have
    FunctionTool --> FunctionToolResult : produces
    
    note for ToolFunction "三种函数签名<br/>根据是否需要上下文<br/>自动选择合适的调用方式"
    note for FunctionToolResult "工具执行结果<br/>包含工具、输出和运行项<br/>用于追踪和调试"