"""
Core tool system providing the foundation for creating and managing tools.

This module defines the base classes and decorators for creating tools in AgentPress:
- Tool base class for implementing tool functionality
- Schema decorators for OpenAPI tool definitions
- Result containers for standardized tool outputs
"""

from typing import Dict, Any, Union, Optional, List
from dataclasses import dataclass, field
from abc import ABC
import json
import inspect
from enum import Enum
from utils.logger import logger

class SchemaType(Enum):
    """Enumeration of supported schema types for tool definitions."""
    OPENAPI = "openapi"
    USAGE_EXAMPLE = "usage_example"

@dataclass
class ToolSchema:
    """Container for tool schemas with type information.
    
    Attributes:
        schema_type (SchemaType): Type of schema (OpenAPI)
        schema (Dict[str, Any]): The actual schema definition
    """
    schema_type: SchemaType
    schema: Dict[str, Any]

@dataclass
class ToolResult:
    """Container for tool execution results.
    
    Attributes:
        success (bool): Whether the tool execution succeeded
        output (str): Output message or error description
    """
    success: bool
    output: str

class Tool(ABC):
    """Abstract base class for all tools.
    
    Provides the foundation for implementing tools with schema registration
    and result handling capabilities.
    
    Attributes:
        _schemas (Dict[str, List[ToolSchema]]): Registered schemas for tool methods
        
    Methods:
        get_schemas: Get all registered tool schemas
        success_response: Create a successful result
        fail_response: Create a failed result
    """
    
    def __init__(self):
        """Initialize tool with empty schema registry."""
        self._schemas: Dict[str, List[ToolSchema]] = {}
        logger.debug(f"Initializing tool class: {self.__class__.__name__}")
        self._register_schemas()

    def _register_schemas(self):
        """Register schemas from all decorated methods."""
        for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
            if hasattr(method, 'tool_schemas'):
                self._schemas[name] = method.tool_schemas
                logger.debug(f"Registered schemas for method '{name}' in {self.__class__.__name__}")

    def get_schemas(self) -> Dict[str, List[ToolSchema]]:
        """Get all registered tool schemas.
        
        Returns:
            Dict mapping method names to their schema definitions
        """
        return self._schemas

    def success_response(self, data: Union[Dict[str, Any], str]) -> ToolResult:
        """Create a successful tool result.
        
        Args:
            data: Result data (dictionary or string)
            
        Returns:
            ToolResult with success=True and formatted output
        """
        if isinstance(data, str):
            text = data
        else:
            text = json.dumps(data, indent=2)
        logger.debug(f"Created success response for {self.__class__.__name__}")
        return ToolResult(success=True, output=text)

    def fail_response(self, msg: str) -> ToolResult:
        """Create a failed tool result.
        
        Args:
            msg: Error message describing the failure
            
        Returns:
            ToolResult with success=False and error message
        """
        logger.debug(f"Tool {self.__class__.__name__} returned failed result: {msg}")
        return ToolResult(success=False, output=msg)

def _add_schema(func, schema: ToolSchema):
    """Helper to add schema to a function."""
    if not hasattr(func, 'tool_schemas'):
        func.tool_schemas = []
    func.tool_schemas.append(schema)
    logger.debug(f"Added {schema.schema_type.value} schema to function {func.__name__}")
    return func

def openapi_schema(schema: Dict[str, Any]):
    """Decorator for OpenAPI schema tools."""
    def decorator(func):
        logger.debug(f"Applying OpenAPI schema to function {func.__name__}")
        return _add_schema(func, ToolSchema(
            schema_type=SchemaType.OPENAPI,
            schema=schema
        ))
    return decorator

def usage_example(example: str):
    """Decorator for providing usage examples for tools in prompts."""
    def decorator(func):
        logger.debug(f"Adding usage example to function {func.__name__}")
        return _add_schema(func, ToolSchema(
            schema_type=SchemaType.USAGE_EXAMPLE,
            schema={"example": example}
        ))
    return decorator

# def xml_schema(**kwargs):
#     """Deprecated decorator - does nothing, kept for compatibility."""
#     def decorator(func):
#         logger.debug(f"xml_schema decorator called on {func.__name__} - ignoring (deprecated)")
#         return func
#     return decorator

MessageTool 是一个用于用户交互的工具类，在整个项目中扮演着与用户沟通的关键角色。

用户沟通桥梁，提供 ask 方法向用户提问并等待回复，

text: str
这是要向用户提出的问题文本
应该是清晰、具体的，并包含必要的上下文信息
attachments: Optional[Union[str, List[str]]] = None
可选参数，可以附加文件或URL到问题中
支持单个字符串或字符串列表
文件路径应相对于 /workspace 目录

if attachments and isinstance(attachments, str):
    attachments = [attachments]

统一处理为列表形式

提供 web_browser_takeover 方法请求用户接管浏览器操作，提供 complete 方法标记任务完成状态。所有方法都是异步实现，支持在异步环境中运行。

当代理需要澄清需求、确认操作或获取信息时使用 ask 方法
遇到需要人工干预的网页操作时使用 web_browser_takeover
完成所有任务后使用 complete 方法通知系统和用户

任务状态管理
通过 complete 方法明确标识任务完成状态
为系统提供清晰的任务执行边界

错误处理和异常情况应对
当自动化工具无法处理复杂情况时，通过 web_browser_takeover 请求用户介入
提供详细的错误信息反馈机制

OpenAPI 兼容
使用 openapi_schema 装饰器定义标准的 OpenAPI 函数接口
支持工具的自动发现和调用

class Tool(ABC):
    """Abstract base class for all tools.
    
    Provides the foundation for implementing tools with schema registration
    and result handling capabilities.
    
    Attributes:
        _schemas (Dict[str, List[ToolSchema]]): Registered schemas for tool methods
        
    Methods:
        get_schemas: Get all registered tool schemas
        success_response: Create a successful result
        fail_response: Create a failed result
    """
    
    def __init__(self):
        """Initialize tool with empty schema registry."""
        self._schemas: Dict[str, List[ToolSchema]] = {}
        logger.debug(f"Initializing tool class: {self.__class__.__name__}")
        self._register_schemas()

    def _register_schemas(self):
        """Register schemas from all decorated methods."""
        for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
            if hasattr(method, 'tool_schemas'):
                self._schemas[name] = method.tool_schemas
                logger.debug(f"Registered schemas for method '{name}' in {self.__class__.__name__}")

    def get_schemas(self) -> Dict[str, List[ToolSchema]]:
        """Get all registered tool schemas.
        
        Returns:
            Dict mapping method names to their schema definitions
        """
        return self._schemas

    def success_response(self, data: Union[Dict[str, Any], str]) -> ToolResult:
        """Create a successful tool result.
        
        Args:
            data: Result data (dictionary or string)
            
        Returns:
            ToolResult with success=True and formatted output
        """
        if isinstance(data, str):
            text = data
        else:
            text = json.dumps(data, indent=2)
        logger.debug(f"Created success response for {self.__class__.__name__}")
        return ToolResult(success=True, output=text)

    def fail_response(self, msg: str) -> ToolResult:
        """Create a failed tool result.
        
        Args:
            msg: Error message describing the failure
            
        Returns:
            ToolResult with success=False and error message
        """
        logger.debug(f"Tool {self.__class__.__name__} returned failed result: {msg}")
        return ToolResult(success=False, output=msg)

类定义和继承关系  class Tool(ABC):
 

self._schemas: Dict[str, List[ToolSchema]] = {}

_schemas：存储工具方法的模式定义
键是方法名，值是 ToolSchema 对象列表
用于记录每个工具方法的 OpenAPI 模式和使用示例

def __init__(self):
    self._schemas: Dict[str, List[ToolSchema]] = {}
    logger.debug(f"Initializing tool class: {self.__class__.__name__}")
    self._register_schemas()

初始化 _schemas 字典为空
记录初始化日志
调用 _register_schemas() 方法注册所有装饰过的方法模式

def _register_schemas(self):
    for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
        if hasattr(method, 'tool_schemas'):
            self._schemas[name] = method.tool_schemas

使用 inspect 模块遍历类中的所有方法

使用 inspect.getmembers() 函数获取当前实例的所有成员
predicate=inspect.ismethod 过滤条件确保只获取方法（而不是属性或其他成员）
返回一个元组列表，每个元组包含 (名称, 方法对象)
例如，对于 MessageTool 实例，这将返回类似：

查找具有 tool_schemas 属性的方法（即被装饰器装饰过的方法）
将这些方法的模式注册到 _schemas 中

def get_schemas(self) -> Dict[str, List[ToolSchema]]:
    return self._schemas

提供对外访问已注册模式的方法
返回当前工具的所有方法模式定义

def openapi_schema(schema: Dict[str, Any]):
    # 用于为工具方法添加 OpenAPI 模式定义

 用于为工具方法添加 OpenAPI 模式定义

def openapi_schema(schema: Dict[str, Any]):
    """Decorator for OpenAPI schema tools."""
    def decorator(func):
        logger.debug(f"Applying OpenAPI schema to function {func.__name__}")
        return _add_schema(func, ToolSchema(
            schema_type=SchemaType.OPENAPI,
            schema=schema
        ))
    return decorator

openapi_schema 装饰器用于装饰 Tool 子类中的方法，为其添加 OpenAPI 模式定义。

主要提供商包括：OpenAI (GPT系列) Anthropic (Claude系列) GroqxAI (Grok系列) Google Gemini
AWS Bedrock OpenRouter（作为统一接口）

def setup_api_keys() -> None:
    """Set up API keys from environment variables."""
    providers = ['OPENAI', 'ANTHROPIC', 'GROQ', 'OPENROUTER', 'XAI', 'MORPH', 'GEMINI']
    # 为各种提供商设置API密钥

effort_level = reasoning_effort or 'low'
如果未指定推理努力程度，则默认为'low'级别

is_anthropic = "anthropic" in model_name.lower() or "claude" in model_name.lower()
is_xai = "xai" in model_name.lower() or model_name.startswith("xai/")

通过检查模型名称来识别是否为Anthropic（Claude）或xAI（Grok）模型。

if is_anthropic:
    params["reasoning_effort"] = effort_level
    params["temperature"] = 1.0  # Required by Anthropic when reasoning_effort is used
    logger.info(f"Anthropic thinking enabled with reasoning_effort='{effort_level}'")

设置reasoning_effort参数，控制推理的深度

# 确保沙箱已初始化
await self._ensure_sandbox()  在执行任何浏览器操作之前，首先确保沙箱环境已经准备就绪。

# 构建 curl 命令
url = f"http://localhost:8003/api/automation/{endpoint}"

if method == "GET" and params:
    query_params = "&".join([f"{k}={v}" for k, v in params.items()])
    url = f"{url}?{query_params}"
    curl_cmd = f"curl -s -X {method} '{url}' -H 'Content-Type: application/json'"
else:
    curl_cmd = f"curl -s -X {method} '{url}' -H 'Content-Type: application/json'"
    if params:
        json_data = json.dumps(params)
        curl_cmd += f" -d '{json_data}'"

函数通过构建 curl 命令来与浏览器自动化 API 通信，支持 GET 和 POST 方法。

response = await self.sandbox.process.exec(curl_cmd, timeout=30)
在沙箱环境中执行构建好的 curl 命令，设置 30 秒超时时间

if "screenshot_base64" in result:
    try:
        # 验证 base64 图像数据
        screenshot_data = result["screenshot_base64"]
        is_valid, validation_message = self._validate_base64_image(screenshot_data)
        
        if is_valid:
            # 上传截图到云存储
            image_url = await upload_base64_image(screenshot_data)
            result["image_url"] = image_url
        # 删除原始 base64 数据以保持结果清洁
        del result["screenshot_base64"]
    except Exception as e:
        result["image_upload_error"] = str(e)

如果响应中包含截图数据，函数会验证图像数据并上传到云存储服务，然后将截图 URL 添加到结果中。

added_message = await self.thread_manager.add_message(
    thread_id=self.thread_id,
    type="browser_state",
    content=result,
    is_llm_message=False
)

# 示例：导航到 URL
async def browser_navigate_to(self, url: str) -> ToolResult:
    return await self._execute_browser_action("navigate_to", {"url": url})

# 示例：点击元素
async def browser_click_element(self, index: int) -> ToolResult:
    return await self._execute_browser_action("click_element", {"index": index})

所有具体的浏览器操作（如导航、点击、输入文本等）都通过这个函数执行

# 设置LiteLLM参数
litellm.modify_params = True  # 自动调整参数
litellm.drop_params = True    # 丢弃不支持的参数
MAX_RETRIES = 3               # 最大重试次数
 

setup_api_keys() 函数负责从环境变量中读取各种提供商的API密钥，包括：
OpenAI，Anthropic，Groq，OpenRouter,XAI，MORPH，GEMINI，AWS Bedrock

_configure_token_limits()：配置token限制
_apply_anthropic_caching()：为Anthropic模型应用缓存控制
_configure_anthopic()：配置Anthropic特定参数
_configure_openrouter()：配置OpenRouter特定参数
_configure_bedrock()：配置AWS Bedrock特定参数
_configure_openai_gpt5()：配置OpenAI GPT-5特定参数
_configure_kimi_k2()：配置Kimi K2特定参数
_configure_thinking()：配置推理/思考参数

prepare_params()
准备API调用所需的所有参数，整合所有配置函数的结果。

make_llm_api_call()
这是主要的API调用函数，支持以下参数：
messages：对话消息列表
model_name：模型名称
temperature：采样温度
max_tokens：最大token数
tools：工具定义列表
stream：是否流式输出

# 调用LLM API
response = await make_llm_api_call(
    messages=[{"role": "user", "content": "你好"}],
    model_name="gpt-4",
    temperature=0.7,
    max_tokens=1000
)
 

agentpress/thread_manager.py - 在 _run_once 函数中调用 LLM 处理对话：

   response = await make_llm_api_call(
       messages=messages,
       model_name=llm_model,
       temperature=llm_temperature,
       max_tokens=llm_max_tokens,
       tools=tools,
       tool_choice=tool_choice,
       stream=True,
       top_p=0.95,
       enable_thinking=enable_thinking,
       reasoning_effort=reasoning_effort
   )

   run    AgentRunner 类

def __init__(self, config: AgentConfig):
    self.config = config

AgentRunner 接受一个 AgentConfig 对象作为参数，该对象包含运行代理所需的所有配置信息：
thread_id: 会话线程ID，project_id: 项目ID
stream: 是否使用流式响应，model_name: 使用的LLM模型
enable_thinking: 是否启用推理模式,agent_config: 代理配置

async def setup(self):setup 方法负责初始化运行环境：
跟踪系统初始化： 如果没有提供跟踪对象，创建一个新的 Langfuse 跟踪实例
线程管理器初始化： 创建 ThreadManager 实例，负责管理对话线程和工具执行
数据库连接： 获取数据库客户端 获取账户ID
项目验证： 验证项目是否存在 检查沙盒配置

async def setup_mcp_tools(self) -> Optional[MCPToolWrapper]:
setup_mcp_tools 方法负责设置 MCP (Model Context Protocol) 工具：
创建 MCPManager 实例 注册 MCP 工具：
解析配置中的 MCP 服务器 处理不同类型的 MCP 连接（pipedream, composio等）
初始化 MCP 工具包装器 将工具注册到工具注册表中

system_message = await PromptManager.build_system_prompt(
    self.config.model_name, self.config.agent_config, 
    self.config.is_agent_builder, self.config.thread_id, 
    mcp_wrapper_instance, self.client
)

使用 PromptManager 构建系统提示，包含：
默认系统提示内容
代理构建器提示（如果是代理构建器）
代理知识库上下文
MCP 工具信息
当前日期和时间信息

主执行循环

主循环最多执行 max_iterations 次（默认100次），在每次迭代中：
账单检查：

检查用户是否有权限继续执行
消息检查：

检查最新消息类型，如果已经是助手消息则停止执行
构建临时消息：

处理浏览器状态、图像上下文等临时信息
调用线程管理器：

这是核心调用，通过 ThreadManager 执行代理逻辑
处理响应流：

处理来自 LLM 的流式响应，检查终止条件
终止条件检查：

检查是否应该终止执行（如调用了 ask、complete 等工具）

action ai筛选过后的提示词 query生成再拿来喂给ai  action input入参

thought 总结搜索结果   final 接口