from abc import ABC, abstractmethod
from typing import Any

class BaseTool(ABC):
    """Framework's tool abstraction."""
    name: str
    description: str

    @abstractmethod
    def execute(self, **kwargs) -> Any:
        """Execute the tool with validated arguments."""
        ...

    @property
    @abstractmethod
    def parameters_schema(self) -> dict:
        """Return JSON Schema for parameters."""
        ...

from typing import Protocol, runtime_checkable

@runtime_checkable
class Tool(Protocol):
    """Structural typing for tools."""
    name: str
    description: str

    def __call__(self, **kwargs) -> Any: ...
    def get_schema(self) -> dict: ...

from functools import wraps

def tool(name: str = None, description: str = None):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)
        wrapper._tool_name = name or func.__name__
        wrapper._tool_description = description or func.__doc__
        wrapper._is_tool = True
        return wrapper
    return decorator

@tool(description="Search the web for information")
def search(query: str, max_results: int = 10) -> list[str]:
    ...

from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    """Input schema for search tool."""
    query: str = Field(description="The search query")
    max_results: int = Field(default=10, ge=1, le=100)

class SearchTool(BaseTool):
    name = "search"
    description = "Search the web"
    args_schema = SearchInput

    def execute(self, **kwargs) -> list[str]:
        validated = SearchInput(**kwargs)
        return perform_search(validated.query, validated.max_results)

import inspect
from typing import get_type_hints

def generate_schema_from_function(func) -> dict:
    hints = get_type_hints(func)
    sig = inspect.signature(func)
    doc = inspect.getdoc(func) or ""

    schema = {
        "type": "function",
        "function": {
            "name": func.__name__,
            "description": doc.split("\n")[0],
            "parameters": {
                "type": "object",
                "properties": {},
                "required": []
            }
        }
    }

    for name, param in sig.parameters.items():
        if name in ("self", "cls"):
            continue

        prop = {"type": python_type_to_json(hints.get(name, str))}

        # Extract description from docstring if available
        if f":param {name}:" in doc:
            prop["description"] = extract_param_doc(doc, name)

        if param.default is inspect.Parameter.empty:
            schema["function"]["parameters"]["required"].append(name)
        else:
            prop["default"] = param.default

        schema["function"]["parameters"]["properties"][name] = prop

    return schema

from pydantic import BaseModel, Field
from pydantic.json_schema import GenerateJsonSchema

class SearchInput(BaseModel):
    """Search the web for information."""
    query: str = Field(description="The search query")
    max_results: int = Field(default=10, ge=1, le=100, description="Max results to return")

def generate_schema_from_pydantic(model: type[BaseModel]) -> dict:
    return {
        "type": "function",
        "function": {
            "name": model.__name__.replace("Input", "").lower(),
            "description": model.__doc__ or "",
            "parameters": model.model_json_schema()
        }
    }

@tool(
    name="search",
    description="Search the web for information",
    parameters={
        "query": {"type": "string", "description": "Search query"},
        "max_results": {"type": "integer", "default": 10}
    }
)
def search(query: str, max_results: int = 10) -> list[str]:
    ...

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "search",
            "description": "Search the web for information",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "The search query"
                    }
                },
                "required": ["query"]
            }
        }
    }
]

agent = Agent(
    tools=[search_tool, calculator_tool, weather_tool]
)

TOOL_REGISTRY = {}

def register_tool(name: str = None):
    def decorator(func):
        tool_name = name or func.__name__
        TOOL_REGISTRY[tool_name] = func
        return func
    return decorator

@register_tool("search")
def search(query: str) -> list[str]: ...

# Agent uses registry
agent = Agent(tools=list(TOOL_REGISTRY.values()))

import importlib
import pkgutil

def discover_tools(package):
    tools = []
    for module_info in pkgutil.iter_modules(package.__path__):
        module = importlib.import_module(f"{package.__name__}.{module_info.name}")
        for name, obj in inspect.getmembers(module):
            if hasattr(obj, '__tool__') or isinstance(obj, BaseTool):
                tools.append(obj)
    return tools

# Usage
from myapp import tools as tools_package
agent = Agent(tools=discover_tools(tools_package))

class ToolFactory:
    _registry: dict[str, type[BaseTool]] = {}

    @classmethod
    def register(cls, name: str):
        def decorator(tool_class):
            cls._registry[name] = tool_class
            return tool_class
        return decorator

    @classmethod
    def create(cls, config: ToolConfig) -> BaseTool:
        tool_class = cls._registry.get(config.type)
        if not tool_class:
            raise ValueError(f"Unknown tool type: {config.type}")
        return tool_class(**config.params)

# Registration
@ToolFactory.register("search")
class SearchTool(BaseTool): ...

# Creation
tool = ToolFactory.create(ToolConfig(type="search", params={"api_key": "..."}))

class WebToolkit:
    """Collection of related tools."""

    def __init__(self, api_key: str):
        self.api_key = api_key

    def get_tools(self) -> list[BaseTool]:
        return [
            SearchTool(api_key=self.api_key),
            BrowseTool(api_key=self.api_key),
            ExtractTool(api_key=self.api_key)
        ]

# Usage
agent = Agent(tools=WebToolkit(api_key="...").get_tools())

def run_tool(self, tool, args):
    try:
        return tool.execute(**args)
    except Exception:
        return None  # Error lost - LLM has no feedback

def run_tool(self, tool, args):
    try:
        return tool.execute(**args)
    except Exception as e:
        return f"Error: {type(e).__name__}"

@dataclass
class ToolResult:
    success: bool
    output: Any = None
    error: str | None = None
    error_type: str | None = None
    suggestion: str | None = None

def run_tool(self, tool, args) -> ToolResult:
    try:
        # Validate first
        validated = tool.validate_args(args)
        result = tool.execute(**validated)
        return ToolResult(success=True, output=result)
    except ValidationError as e:
        return ToolResult(
            success=False,
            error=str(e),
            error_type="validation_error",
            suggestion=f"Check parameter types: {e.errors()}"
        )
    except ToolExecutionError as e:
        return ToolResult(
            success=False,
            error=str(e),
            error_type="execution_error",
            suggestion=e.suggestion if hasattr(e, 'suggestion') else None
        )

def format_error_for_llm(self, result: ToolResult) -> str:
    if result.success:
        return str(result.output)

    parts = [f"Tool execution failed: {result.error}"]

    if result.error_type == "validation_error":
        parts.append("The provided arguments did not match the expected schema.")

    if result.suggestion:
        parts.append(f"Suggestion: {result.suggestion}")

    return "\n".join(parts)

async def run_with_retry(self, tool, args, max_retries=3):
    for attempt in range(max_retries):
        result = await self.run_tool(tool, args)
        if result.success:
            return result
        if not self._is_retryable(result.error_type):
            return result
        await asyncio.sleep(2 ** attempt)  # Exponential backoff
    return result

async def run_with_self_correction(self, tool, args, max_retries=3):
    for attempt in range(max_retries):
        result = await self.run_tool(tool, args)
        if result.success:
            return result

        # Ask LLM to fix the error
        correction_prompt = f"""
Tool `{tool.name}` failed with error: {result.error}
Original arguments: {json.dumps(args)}
Tool schema: {json.dumps(tool.parameters_schema)}

Provide corrected arguments as JSON.
"""
        corrected = await self.llm.generate(correction_prompt)
        args = json.loads(corrected)

    return result

async def run_with_fallback(self, tool_chain: list[BaseTool], args):
    for tool in tool_chain:
        result = await self.run_tool(tool, args)
        if result.success:
            return result
    return result  # Return last failure

forensics-output/frameworks/{framework}/phase2/tool-interface-analysis.md

# Tool Interface Analysis: {Framework Name}

## Summary
- **Tool Modeling**: [Base class / Protocol / Decorated functions / Pydantic models]
- **Schema Generation**: [Introspection / Pydantic / Decorator / Manual]
- **Registration Pattern**: [Declarative / Registry / Discovery / Factory]
- **Error Handling**: [Silent / Basic / Detailed / Structured]
- **Built-in Tools**: [Count] tools in [N] categories

## Tool Modeling

### Core Abstraction

**Type**: [Abstract Base Class / Protocol / Decorated Function / Pydantic Model / Hybrid]

**Location**: `path/to/tool.py:L##`

```python
# Show the core tool type definition

BaseTool
├── BuiltinTool
├── APITool
└── CustomTool

# Example code

# Example code

# Show how schemas are generated

{
  "type": "function",
  "function": {
    "name": "example_tool",
    "description": "...",
    "parameters": {...}
  }
}

# Show input type/schema

1. Tool defined →
2. [Registration step] →
3. [Discovery step] →
4. Available to agent

# Show registration code

# Show tool execution code

# Show how errors are formatted for LLM

# Show parallel execution code if present

---

## Integration Points

- **Prerequisite**: `codebase-mapping` to identify tool-related files
- **Related**: `harness-model-protocol` for wire encoding of tool calls
- **Related**: `resilience-analysis` for error handling patterns
- **Feeds into**: `comparative-matrix` for interface decisions
- **Feeds into**: `architecture-synthesis` for tool layer design

## Key Questions to Answer

1. How is a "tool" represented in this framework? (type, class, protocol)
2. How are tool schemas generated from definitions?
3. What built-in tools ship with the framework?
4. How are tools registered and discovered?
5. How is tool execution orchestrated?
6. How are errors fed back to the LLM for retry?
7. Does the framework support parallel tool execution?
8. How does validation work (pre-execution, schema-based)?
9. What retry/self-correction mechanisms exist?
10. Can tools be dynamically added/removed at runtime?

## Files to Examine

When analyzing a framework, prioritize these file patterns:

| Pattern | Purpose |
|---------|---------|
| `**/tool*.py`, `**/tools/**` | Tool definitions and base classes |
| `**/schema*.py` | Schema generation |
| `**/registry*.py`, `**/register*.py` | Tool registration |
| `**/executor*.py`, `**/runner*.py` | Tool execution |
| `**/builtin*.py`, `**/default*.py` | Built-in tool inventory |
| `**/error*.py`, `**/exception*.py` | Error types and handling |
| `**/validation*.py` | Argument validation |
| `**/function*.py`, `**/callable*.py` | Function-based tools |