Tool Executor in Machine Learning

import openai
import json
from typing import Any, Callable
from dataclasses import dataclass, field
import time
import logging

logger = logging.getLogger(__name__)

@dataclass
class Tool:
    name: str
    description: str
    parameters: dict  # JSON Schema
    function: Callable
    timeout: float = 30.0
    requires_confirmation: bool = False

class ToolExecutor:
    def __init__(self):
        self.tools: dict[str, Tool] = {}
        self.call_log: list[dict] = []

    def register(self, tool: Tool):
        self.tools[tool.name] = tool

    def get_openai_tools(self) -> list[dict]:
        """Convert registered tools to OpenAI function calling format."""
        return [
            {
                "type": "function",
                "function": {
                    "name": t.name,
                    "description": t.description,
                    "parameters": t.parameters,
                },
            }
            for t in self.tools.values()
        ]

    def execute(self, tool_call) -> dict[str, Any]:
        """Execute a single tool call with validation, timeout, and logging."""
        name = tool_call.function.name
        start_time = time.time()
        log_entry = {"tool": name, "timestamp": start_time}

        # Step 1: Check tool exists
        if name not in self.tools:
            error = f"Unknown tool: {name}. Available: {list(self.tools.keys())}"
            log_entry["error"] = error
            self.call_log.append(log_entry)
            return {"error": error}

        tool = self.tools[name]

        # Step 2: Parse arguments
        try:
            args = json.loads(tool_call.function.arguments)
        except json.JSONDecodeError as e:
            error = f"Invalid JSON arguments: {e}"
            log_entry["error"] = error
            self.call_log.append(log_entry)
            return {"error": error}

        log_entry["args"] = args

        # Step 3: Execute with timeout
        try:
            import concurrent.futures
            with concurrent.futures.ThreadPoolExecutor() as pool:
                future = pool.submit(tool.function, **args)
                result = future.result(timeout=tool.timeout)

            log_entry["result"] = result
            log_entry["latency_ms"] = (time.time() - start_time) * 1000
            self.call_log.append(log_entry)
            return {"result": result}

        except concurrent.futures.TimeoutError:
            error = f"Tool '{name}' timed out after {tool.timeout}s"
            log_entry["error"] = error
            self.call_log.append(log_entry)
            return {"error": error}
        except Exception as e:
            error = f"Tool '{name}' failed: {type(e).__name__}: {e}"
            log_entry["error"] = error
            self.call_log.append(log_entry)
            return {"error": error}

    def run_agent_loop(
        self, client: openai.OpenAI, messages: list, model: str = "gpt-4o"
    ) -> str:
        """Run the full agent loop until the model stops calling tools."""
        tools = self.get_openai_tools()
        max_iterations = 10  # Safety limit

        for _ in range(max_iterations):
            response = client.chat.completions.create(
                model=model, messages=messages, tools=tools
            )
            msg = response.choices[0].message
            messages.append(msg)

            if not msg.tool_calls:
                return msg.content  # Final text response

            # Execute all tool calls (potentially in parallel)
            for tc in msg.tool_calls:
                result = self.execute(tc)
                messages.append(
                    {
                        "role": "tool",
                        "tool_call_id": tc.id,
                        "content": json.dumps(result),
                    }
                )

        return "Agent reached maximum iteration limit."


# --- Usage Example ---
def get_weather(city: str, units: str = "celsius") -> dict:
    """Simulated weather API call."""
    # In production, this would call a real weather API
    return {"city": city, "temperature": 32, "units": units, "condition": "sunny"}

def search_products(query: str, max_results: int = 5) -> list:
    """Simulated product search."""
    return [{"name": f"Result for '{query}'", "price": 499, "currency": "INR"}]


executor = ToolExecutor()

executor.register(Tool(
    name="get_weather",
    description="Get current weather for a city",
    parameters={
        "type": "object",
        "properties": {
            "city": {"type": "string", "description": "City name"},
            "units": {"type": "string", "enum": ["celsius", "fahrenheit"]},
        },
        "required": ["city"],
    },
    function=get_weather,
    timeout=10.0,
))

executor.register(Tool(
    name="search_products",
    description="Search for products by keyword",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "Search query"},
            "max_results": {"type": "integer", "default": 5},
        },
        "required": ["query"],
    },
    function=search_products,
    timeout=15.0,
))

client = openai.OpenAI()
result = executor.run_agent_loop(
    client,
    messages=[{"role": "user", "content": "What's the weather in Mumbai?"}],
)

import anthropic
import json
from jsonschema import validate, ValidationError

client = anthropic.Anthropic()

# Define tools with JSON Schema
tools = [
    {
        "name": "lookup_order",
        "description": "Look up an order by order ID. Returns order details including status, items, and total amount in INR.",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "string",
                    "pattern": "^ORD-[0-9]{6}$",
                    "description": "Order ID in format ORD-XXXXXX",
                },
                "include_items": {
                    "type": "boolean",
                    "default": True,
                    "description": "Whether to include line items",
                },
            },
            "required": ["order_id"],
        },
    },
    {
        "name": "process_refund",
        "description": "Process a refund for an order. Amount is in INR. Requires reason.",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string", "pattern": "^ORD-[0-9]{6}$"},
                "amount": {"type": "number", "minimum": 1, "maximum": 100000},
                "reason": {
                    "type": "string",
                    "enum": ["defective", "wrong_item", "not_delivered", "other"],
                },
            },
            "required": ["order_id", "amount", "reason"],
        },
    },
]


def execute_tool(name: str, input_args: dict) -> str:
    """Execute a tool with pre-validation."""
    # Find the tool schema
    tool_def = next((t for t in tools if t["name"] == name), None)
    if not tool_def:
        return json.dumps({"error": f"Unknown tool: {name}"})

    # Validate input against schema
    try:
        validate(instance=input_args, schema=tool_def["input_schema"])
    except ValidationError as e:
        return json.dumps({"error": f"Validation failed: {e.message}"})

    # Dispatch to implementation
    if name == "lookup_order":
        return json.dumps({
            "order_id": input_args["order_id"],
            "status": "delivered",
            "total": 2499,
            "currency": "INR",
            "items": [{"name": "Wireless Earbuds", "qty": 1, "price": 2499}],
        })
    elif name == "process_refund":
        return json.dumps({
            "refund_id": "REF-789012",
            "status": "processed",
            "amount": input_args["amount"],
            "currency": "INR",
        })
    return json.dumps({"error": "Not implemented"})


def run_claude_agent(user_message: str) -> str:
    """Run a tool-augmented conversation with Claude."""
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            tools=tools,
            messages=messages,
        )

        # Collect text and tool_use blocks
        assistant_content = response.content
        messages.append({"role": "assistant", "content": assistant_content})

        # Check if we need to execute tools
        tool_blocks = [b for b in assistant_content if b.type == "tool_use"]

        if response.stop_reason == "end_turn" or not tool_blocks:
            # Extract final text
            text_blocks = [b.text for b in assistant_content if b.type == "text"]
            return "\n".join(text_blocks)

        # Execute each tool and return results
        tool_results = []
        for block in tool_blocks:
            result = execute_tool(block.name, block.input)
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": result,
            })

        messages.append({"role": "user", "content": tool_results})


# Example: Customer service agent
response = run_claude_agent(
    "I received the wrong item in order ORD-123456. I'd like a refund."
)
print(response)

from mcp.server.fastmcp import FastMCP
import httpx
import json

# Create an MCP server that exposes tools
mcp = FastMCP("ecommerce-tools")


@mcp.tool()
async def check_inventory(
    product_id: str,
    warehouse: str = "default",
) -> str:
    """Check real-time inventory for a product.

    Args:
        product_id: The product SKU or ID
        warehouse: Warehouse code (default, mumbai, delhi, bengaluru)

    Returns:
        JSON with product availability and stock count
    """
    # In production, query your inventory database
    inventory_data = {
        "product_id": product_id,
        "warehouse": warehouse,
        "in_stock": True,
        "quantity": 142,
        "last_updated": "2026-02-07T10:30:00Z",
    }
    return json.dumps(inventory_data)


@mcp.tool()
async def create_payment_link(
    amount: float,
    currency: str = "INR",
    description: str = "",
    customer_email: str = "",
) -> str:
    """Create a payment link using Razorpay.

    Args:
        amount: Payment amount (minimum 1.00)
        currency: Currency code (INR, USD)
        description: Payment description shown to customer
        customer_email: Customer's email for receipt

    Returns:
        JSON with payment link URL and link ID
    """
    if amount < 1.0:
        return json.dumps({"error": "Amount must be at least 1.00"})

    # In production, call Razorpay API
    # async with httpx.AsyncClient() as client:
    #     response = await client.post(
    #         "https://api.razorpay.com/v1/payment_links",
    #         auth=(RAZORPAY_KEY_ID, RAZORPAY_KEY_SECRET),
    #         json={"amount": int(amount * 100), "currency": currency, ...}
    #     )
    return json.dumps({
        "link_id": "plink_ExjpAUN3gVHrPJ",
        "short_url": "https://rzp.io/i/example",
        "amount": amount,
        "currency": currency,
        "status": "created",
    })


@mcp.tool()
async def query_analytics(
    metric: str,
    time_range: str = "7d",
    group_by: str = "day",
) -> str:
    """Query business analytics metrics.

    Args:
        metric: Metric name (revenue, orders, aov, conversion_rate)
        time_range: Time range (1d, 7d, 30d, 90d)
        group_by: Grouping (hour, day, week, month)

    Returns:
        JSON with metric values and trend data
    """
    return json.dumps({
        "metric": metric,
        "time_range": time_range,
        "value": 1_45_000,  # Indian numbering: 1,45,000 INR
        "currency": "INR",
        "trend": "+12.5%",
        "data_points": [
            {"date": "2026-02-01", "value": 18500},
            {"date": "2026-02-02", "value": 21000},
            {"date": "2026-02-03", "value": 19200},
        ],
    })


if __name__ == "__main__":
    mcp.run(transport="stdio")

import subprocess
import tempfile
import os
import json
from pathlib import Path


class SandboxedCodeExecutor:
    """Execute LLM-generated Python code in an isolated environment.

    Uses subprocess with restricted permissions. For production,
    replace with E2B, Firecracker microVM, or gVisor.
    """

    def __init__(
        self,
        timeout: int = 30,
        max_output_chars: int = 10000,
        allowed_imports: set[str] | None = None,
    ):
        self.timeout = timeout
        self.max_output_chars = max_output_chars
        self.allowed_imports = allowed_imports or {
            "math", "json", "datetime", "re", "statistics",
            "collections", "itertools", "functools",
            "csv", "io", "textwrap", "decimal", "fractions",
        }

    def validate_code(self, code: str) -> list[str]:
        """Check for disallowed operations before execution."""
        violations = []
        dangerous_patterns = [
            ("import os", "os module access"),
            ("import subprocess", "subprocess access"),
            ("import sys", "sys module access"),
            ("__import__", "dynamic imports"),
            ("eval(", "eval execution"),
            ("exec(", "exec execution"),
            ("open(", "file system access"),
            ("requests.", "network access"),
            ("urllib", "network access"),
            ("socket", "network access"),
        ]
        for pattern, reason in dangerous_patterns:
            if pattern in code:
                violations.append(f"Blocked: {reason} ({pattern})")
        return violations

    def execute(self, code: str) -> dict:
        """Execute code in a sandboxed subprocess."""
        # Pre-execution validation
        violations = self.validate_code(code)
        if violations:
            return {
                "success": False,
                "error": f"Security violations: {'; '.join(violations)}",
                "output": "",
            }

        # Write code to temp file
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".py", delete=False
        ) as f:
            f.write(code)
            temp_path = f.name

        try:
            result = subprocess.run(
                ["python3", temp_path],
                capture_output=True,
                text=True,
                timeout=self.timeout,
                env={"PATH": "/usr/bin:/usr/local/bin"},  # Minimal env
                cwd=tempfile.gettempdir(),
            )

            output = result.stdout[:self.max_output_chars]
            error = result.stderr[:self.max_output_chars]

            return {
                "success": result.returncode == 0,
                "output": output,
                "error": error if result.returncode != 0 else "",
                "return_code": result.returncode,
            }

        except subprocess.TimeoutExpired:
            return {
                "success": False,
                "error": f"Execution timed out after {self.timeout}s",
                "output": "",
            }
        finally:
            os.unlink(temp_path)


# Register as a tool
def run_python_code(code: str) -> dict:
    """Execute Python code in a sandboxed environment."""
    executor = SandboxedCodeExecutor(timeout=30)
    return executor.execute(code)


# Example: LLM generates code to analyze data
result = run_python_code("""
import statistics
data = [45, 67, 23, 89, 56, 78, 34, 91, 12, 65]
print(f"Mean: {statistics.mean(data):.2f}")
print(f"Median: {statistics.median(data):.2f}")
print(f"Std Dev: {statistics.stdev(data):.2f}")
""")
print(result)

# Tool Executor Configuration (YAML)
executor:
  max_parallel_calls: 5
  global_timeout_seconds: 120
  max_iterations: 10          # Max tool-calling rounds per request
  max_result_tokens: 4000     # Truncate results exceeding this

tool_defaults:
  timeout_seconds: 30
  retry_count: 2
  retry_backoff_ms: 1000

sandbox:
  enabled: true
  runtime: "e2b"              # Options: e2b, gvisor, docker, subprocess
  max_execution_time: 30
  max_memory_mb: 512
  network_access: false
  allowed_packages:
    - numpy
    - pandas
    - matplotlib
    - scikit-learn

observability:
  log_tool_calls: true
  log_arguments: true         # Set false if args contain PII
  log_results: true
  metrics_backend: prometheus
  trace_backend: opentelemetry

rate_limiting:
  per_user_rpm: 60            # Requests per minute per user
  per_tool_rpm:
    process_payment: 10
    send_email: 20
    search_products: 100

permissions:
  default_role: "viewer"
  roles:
    viewer:
      allowed_tools: ["search_products", "check_inventory", "get_weather"]
    operator:
      allowed_tools: ["*"]
      denied_tools: ["process_refund"]
    admin:
      allowed_tools: ["*"]