Uso de herramientas, function calling y MCP

1. Developer defines tools (name, description, parameters)
2. Developer sends user message + tool definitions to LLM API
3. LLM decides whether to call a tool
   a. If yes: returns a tool_call with function name + arguments
   b. If no: returns a regular text response
4. Developer executes the tool and sends the result back
5. LLM incorporates the result and continues

"""Complete function calling example with OpenAI API."""

import json
from openai import OpenAI

client = OpenAI()

# Step 1: Define the tools
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": (
                "Get the current weather for a specific location. "
                "Use this when the user asks about weather conditions."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "City name, e.g., 'London' or 'Tokyo, Japan'"
                    },
                    "units": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature unit preference"
                    }
                },
                "required": ["location"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_restaurants",
            "description": (
                "Search for restaurants near a location. "
                "Returns a list of restaurants with ratings and cuisine types."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "Location to search near"
                    },
                    "cuisine": {
                        "type": "string",
                        "description": "Type of cuisine (e.g., 'italian', 'japanese', 'mexican')"
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "Maximum number of results to return",
                        "default": 5
                    }
                },
                "required": ["location"]
            }
        }
    }
]

# Step 2: Send the request
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {
            "role": "system",
            "content": "You help users plan their travel. Use tools when needed."
        },
        {
            "role": "user",
            "content": "What's the weather like in Barcelona, and can you find good tapas places there?"
        }
    ],
    tools=tools,
    tool_choice="auto",  # "auto", "none", or {"type": "function", "function": {"name": "..."}}
)

message = response.choices[0].message

# Step 3: Process tool calls
if message.tool_calls:
    for tool_call in message.tool_calls:
        print(f"Tool: {tool_call.function.name}")
        print(f"Args: {tool_call.function.arguments}")
        print(f"ID:   {tool_call.id}")
        print()

"""Function calling with the Anthropic API."""

import anthropic
import json

client = anthropic.Anthropic()

# Define tools using Anthropic's format
tools = [
    {
        "name": "get_weather",
        "description": (
            "Get the current weather for a specific location. "
            "Returns temperature, conditions, humidity, and wind speed."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "City name, e.g., 'London' or 'Paris, France'"
                },
                "units": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Temperature unit"
                }
            },
            "required": ["location"]
        }
    },
    {
        "name": "calculate",
        "description": "Perform a mathematical calculation. Accepts any valid mathematical expression.",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "Mathematical expression to evaluate, e.g., '(15 * 23) + 7'"
                }
            },
            "required": ["expression"]
        }
    }
]

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system="You are a helpful assistant with access to tools.",
    tools=tools,
    messages=[
        {
            "role": "user",
            "content": "What's the weather in Madrid, and if it's above 25 degrees, how much is that in Fahrenheit?"
        }
    ]
)

# Process the response
for block in response.content:
    if block.type == "text":
        print(f"Text: {block.text}")
    elif block.type == "tool_use":
        print(f"Tool call: {block.name}")
        print(f"Input: {json.dumps(block.input, indent=2)}")
        print(f"ID: {block.id}")

"""Full tool call lifecycle: request → tool call → execution → response."""

import json
from openai import OpenAI

client = OpenAI()

# --- Simulated tool implementations ---
def get_weather(location: str, units: str = "celsius") -> dict:
    """Simulated weather API."""
    weather_data = {
        "Barcelona": {"temp": 22, "condition": "Sunny", "humidity": 65},
        "London": {"temp": 14, "condition": "Cloudy", "humidity": 80},
        "Tokyo": {"temp": 18, "condition": "Rainy", "humidity": 75},
    }
    city = location.split(",")[0].strip()
    data = weather_data.get(city, {"temp": 20, "condition": "Unknown", "humidity": 50})

    if units == "fahrenheit":
        data["temp"] = data["temp"] * 9/5 + 32
        data["unit"] = "°F"
    else:
        data["unit"] = "°C"

    return {"location": location, **data}


def search_restaurants(location: str, cuisine: str = None, max_results: int = 5) -> dict:
    """Simulated restaurant search."""
    results = [
        {"name": "Cal Pep", "cuisine": "tapas", "rating": 4.6, "price": "€€€"},
        {"name": "Bar Cañete", "cuisine": "tapas", "rating": 4.5, "price": "€€"},
        {"name": "Cervecería Catalana", "cuisine": "tapas", "rating": 4.4, "price": "€€"},
    ]
    if cuisine:
        results = [r for r in results if r["cuisine"].lower() == cuisine.lower()]
    return {"results": results[:max_results]}


TOOL_FUNCTIONS = {
    "get_weather": get_weather,
    "search_restaurants": search_restaurants,
}

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a location.",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string", "description": "City name"},
                    "units": {"type": "string", "enum": ["celsius", "fahrenheit"]}
                },
                "required": ["location"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_restaurants",
            "description": "Search for restaurants near a location.",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string"},
                    "cuisine": {"type": "string"},
                    "max_results": {"type": "integer", "default": 5}
                },
                "required": ["location"]
            }
        }
    }
]


def run_agent_with_tools(user_message: str) -> str:
    """Run an agent that can use tools, handling the full lifecycle."""

    messages = [
        {"role": "system", "content": "You help users plan activities. Use tools when helpful."},
        {"role": "user", "content": user_message}
    ]

    max_iterations = 10

    for i in range(max_iterations):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto",
        )

        assistant_message = response.choices[0].message
        messages.append(assistant_message)

        if not assistant_message.tool_calls:
            return assistant_message.content

        for tool_call in assistant_message.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)

            print(f"  [{i+1}] Calling {func_name}({func_args})")

            if func_name in TOOL_FUNCTIONS:
                result = TOOL_FUNCTIONS[func_name](**func_args)
                result_str = json.dumps(result)
            else:
                result_str = json.dumps({"error": f"Unknown tool: {func_name}"})

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result_str,
            })

    return "Agent reached maximum iterations."


# Run the agent
result = run_agent_with_tools(
    "What's the weather in Barcelona? Also find me some good tapas restaurants there."
)
print(f"\nFinal response:\n{result}")

{
    "name": "search",
    "description": "Search for things"
}

{
    "name": "web_search",
    "description": "Search the internet using a search engine. Returns the top results with titles, URLs, and snippets. Use this when you need current information, facts you are not confident about, or information that may have changed after your training cutoff. Do NOT use this for simple factual questions you are confident about (e.g., 'What is the capital of France?')."
}

"Use this when you need to perform mathematical calculations that
require precision. Do NOT use this for estimates or approximations
that you can handle directly."

{
    "date": {
        "type": "string",
        "description": "Date in ISO 8601 format (YYYY-MM-DD). Example: '2025-03-15'. Must be a valid calendar date."
    }
}

{
    "query": {
        "type": "string",
        "description": "Search query. Maximum 200 characters. Use specific, focused queries rather than broad ones. If the first search does not return useful results, try rephrasing with different keywords."
    }
}

"Returns a JSON object with fields: 'results' (array of search results,
each with 'title', 'url', 'snippet'), 'total_count' (integer),
'query_time_ms' (integer)."

{
    "sort_by": {
        "type": "string",
        "enum": ["relevance", "date", "rating", "price_low_to_high", "price_high_to_low"],
        "description": "How to sort the results"
    }
}

{
    "max_results": {
        "type": "integer",
        "default": 10,
        "minimum": 1,
        "maximum": 100,
        "description": "Number of results to return. Default: 10."
    }
}

open_file, read_line, write_line, close_file, move_cursor, ...

manage_files(operation: "read|write|delete|move|copy|list", ...)

read_file(path) → returns file contents
write_file(path, content) → writes content to file
list_directory(path) → lists directory contents

"""Web search tool implementation."""

import httpx

async def web_search(query: str, max_results: int = 5) -> dict:
    """
    Search the web using a search API.

    In production, you would use:
    - Google Custom Search API
    - Bing Search API
    - SerpAPI
    - Brave Search API
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.search-provider.com/search",
            params={"q": query, "count": max_results},
            headers={"Authorization": f"Bearer {API_KEY}"}
        )
        data = response.json()

    return {
        "results": [
            {
                "title": r["title"],
                "url": r["url"],
                "snippet": r["snippet"]
            }
            for r in data.get("results", [])
        ]
    }

"""Safe calculator tool with sandboxed execution."""

import ast
import operator
import math

# Define allowed operations
SAFE_OPERATORS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.FloorDiv: operator.floordiv,
    ast.Mod: operator.mod,
    ast.Pow: operator.pow,
    ast.USub: operator.neg,
}

SAFE_FUNCTIONS = {
    "abs": abs,
    "round": round,
    "min": min,
    "max": max,
    "sqrt": math.sqrt,
    "log": math.log,
    "log10": math.log10,
    "sin": math.sin,
    "cos": math.cos,
    "tan": math.tan,
    "pi": math.pi,
    "e": math.e,
}


def safe_eval(expression: str) -> float:
    """
    Safely evaluate a mathematical expression.

    Only allows arithmetic operations and safe math functions.
    No access to builtins, imports, or arbitrary code execution.
    """
    try:
        tree = ast.parse(expression, mode='eval')
        return _eval_node(tree.body)
    except (ValueError, TypeError, ZeroDivisionError) as e:
        raise ValueError(f"Calculation error: {e}")
    except Exception as e:
        raise ValueError(f"Invalid expression: {e}")


def _eval_node(node):
    """Recursively evaluate an AST node."""
    if isinstance(node, ast.Constant):
        if isinstance(node.value, (int, float)):
            return node.value
        raise ValueError(f"Unsupported constant type: {type(node.value)}")

    elif isinstance(node, ast.BinOp):
        op_func = SAFE_OPERATORS.get(type(node.op))
        if op_func is None:
            raise ValueError(f"Unsupported operator: {type(node.op).__name__}")
        return op_func(_eval_node(node.left), _eval_node(node.right))

    elif isinstance(node, ast.UnaryOp):
        op_func = SAFE_OPERATORS.get(type(node.op))
        if op_func is None:
            raise ValueError(f"Unsupported operator: {type(node.op).__name__}")
        return op_func(_eval_node(node.operand))

    elif isinstance(node, ast.Call):
        if isinstance(node.func, ast.Name) and node.func.id in SAFE_FUNCTIONS:
            func = SAFE_FUNCTIONS[node.func.id]
            args = [_eval_node(arg) for arg in node.args]
            return func(*args)
        raise ValueError(f"Unsupported function: {ast.dump(node.func)}")

    elif isinstance(node, ast.Name):
        if node.id in SAFE_FUNCTIONS:
            return SAFE_FUNCTIONS[node.id]
        raise ValueError(f"Unknown variable: {node.id}")

    raise ValueError(f"Unsupported expression: {ast.dump(node)}")


# Usage
print(safe_eval("2 ** 10"))          # 1024
print(safe_eval("sqrt(144) + 3"))    # 15.0
print(safe_eval("log10(1000)"))      # 3.0

"""Sandboxed Python code execution tool."""

import subprocess
import tempfile
import os

def execute_python(code: str, timeout: int = 30) -> dict:
    """
    Execute Python code in a sandboxed subprocess.

    WARNING: This is a simplified example. Production systems should use
    proper sandboxing (Docker containers, gVisor, E2B, etc.).
    """
    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=timeout,
            env={
                **os.environ,
            }
        )

        return {
            "stdout": result.stdout,
            "stderr": result.stderr,
            "return_code": result.returncode,
            "success": result.returncode == 0,
        }
    except subprocess.TimeoutExpired:
        return {
            "stdout": "",
            "stderr": f"Execution timed out after {timeout} seconds",
            "return_code": -1,
            "success": False,
        }
    finally:
        os.unlink(temp_path)

"""File operation tools for an agent."""

import os
import json

def read_file(filepath: str, max_lines: int = 1000) -> dict:
    """Read the contents of a file."""
    try:
        filepath = os.path.abspath(filepath)
        if not filepath.startswith(ALLOWED_DIRECTORY):
            return {"error": "Access denied: path outside allowed directory"}

        with open(filepath, 'r') as f:
            lines = f.readlines()[:max_lines]

        return {
            "content": "".join(lines),
            "total_lines": len(lines),
            "truncated": len(lines) == max_lines,
        }
    except FileNotFoundError:
        return {"error": f"File not found: {filepath}"}
    except PermissionError:
        return {"error": f"Permission denied: {filepath}"}


def write_file(filepath: str, content: str) -> dict:
    """Write content to a file."""
    try:
        filepath = os.path.abspath(filepath)
        if not filepath.startswith(ALLOWED_DIRECTORY):
            return {"error": "Access denied: path outside allowed directory"}

        os.makedirs(os.path.dirname(filepath), exist_ok=True)

        with open(filepath, 'w') as f:
            f.write(content)

        return {"success": True, "path": filepath, "bytes_written": len(content)}
    except Exception as e:
        return {"error": str(e)}


def list_directory(path: str) -> dict:
    """List the contents of a directory."""
    try:
        path = os.path.abspath(path)
        if not path.startswith(ALLOWED_DIRECTORY):
            return {"error": "Access denied: path outside allowed directory"}

        entries = []
        for entry in os.scandir(path):
            entries.append({
                "name": entry.name,
                "type": "directory" if entry.is_dir() else "file",
                "size": entry.stat().st_size if entry.is_file() else None,
            })

        return {"path": path, "entries": sorted(entries, key=lambda e: e["name"])}
    except Exception as e:
        return {"error": str(e)}

"""Two-stage tool selection: classify then present relevant tools."""

TOOL_CATEGORIES = {
    "information": ["web_search", "wikipedia_lookup", "news_search"],
    "computation": ["calculator", "python_executor", "statistics"],
    "communication": ["send_email", "send_slack", "create_ticket"],
    "files": ["read_file", "write_file", "list_directory"],
}

async def select_tools(query: str) -> list[dict]:
    """Select relevant tools using a fast classifier model."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",  # Fast, cheap model for classification
        messages=[
            {
                "role": "system",
                "content": (
                    "Classify the user's intent into one or more categories: "
                    "information, computation, communication, files. "
                    "Return ONLY the category names, comma-separated."
                )
            },
            {"role": "user", "content": query}
        ],
        temperature=0.0,
    )

    categories = [c.strip() for c in response.choices[0].message.content.split(",")]

    selected_tools = []
    for category in categories:
        if category in TOOL_CATEGORIES:
            for tool_name in TOOL_CATEGORIES[category]:
                selected_tools.append(TOOL_DEFINITIONS[tool_name])

    return selected_tools

"""Semantic tool routing using embeddings."""

import numpy as np
from openai import OpenAI

client = OpenAI()

def get_embedding(text: str) -> list[float]:
    """Get the embedding for a text using OpenAI's embedding model."""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding


def cosine_similarity(a: list[float], b: list[float]) -> float:
    """Compute cosine similarity between two vectors."""
    a, b = np.array(a), np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))


class ToolRouter:
    """Route queries to relevant tools using semantic similarity."""

    def __init__(self, tools: list[dict]):
        self.tools = tools
        self.tool_embeddings = []
        for tool in tools:
            desc = f"{tool['name']}: {tool['description']}"
            self.tool_embeddings.append(get_embedding(desc))

    def select_tools(self, query: str, top_k: int = 3, threshold: float = 0.3) -> list[dict]:
        """Select the top_k most relevant tools for the query."""
        query_embedding = get_embedding(query)

        scores = []
        for i, tool_emb in enumerate(self.tool_embeddings):
            score = cosine_similarity(query_embedding, tool_emb)
            scores.append((score, i))

        scores.sort(reverse=True)

        selected = []
        for score, idx in scores[:top_k]:
            if score >= threshold:
                selected.append(self.tools[idx])

        return selected

User: "What's the weather in the cheapest flight destination from Madrid?"
1. search_flights(from="Madrid", sort="price") → Result: "Lisbon, €45"
2. get_weather(location="Lisbon") → Result: "22°C, Sunny"

"""Handling parallel tool calls."""

import asyncio
import json

async def execute_tool_calls_parallel(tool_calls: list) -> list[dict]:
    """Execute multiple tool calls in parallel."""

    async def execute_one(tool_call):
        func_name = tool_call.function.name
        func_args = json.loads(tool_call.function.arguments)

        if func_name in ASYNC_TOOL_FUNCTIONS:
            result = await ASYNC_TOOL_FUNCTIONS[func_name](**func_args)
        elif func_name in TOOL_FUNCTIONS:
            result = TOOL_FUNCTIONS[func_name](**func_args)
        else:
            result = {"error": f"Unknown tool: {func_name}"}

        return {
            "tool_call_id": tool_call.id,
            "role": "tool",
            "content": json.dumps(result),
        }

    results = await asyncio.gather(*[execute_one(tc) for tc in tool_calls])
    return results

"""Robust tool execution with error handling and retries."""

import time
import json
import traceback
from functools import wraps

class ToolExecutionError(Exception):
    """Custom exception for tool execution failures."""
    def __init__(self, tool_name: str, error_type: str, message: str, retryable: bool = False):
        self.tool_name = tool_name
        self.error_type = error_type
        self.message = message
        self.retryable = retryable
        super().__init__(message)


def with_retry(max_retries: int = 3, backoff_factor: float = 1.0):
    """Decorator that adds retry logic to a tool function."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except ToolExecutionError as e:
                    if not e.retryable:
                        raise
                    last_error = e
                    wait_time = backoff_factor * (2 ** attempt)
                    print(f"  Retry {attempt + 1}/{max_retries} for {e.tool_name} "
                          f"after {wait_time}s: {e.message}")
                    time.sleep(wait_time)
                except Exception as e:
                    last_error = e
                    wait_time = backoff_factor * (2 ** attempt)
                    print(f"  Retry {attempt + 1}/{max_retries} after {wait_time}s: {e}")
                    time.sleep(wait_time)
            raise last_error
        return wrapper
    return decorator


def execute_tool_safely(tool_name: str, tool_func, args: dict) -> str:
    """Execute a tool with comprehensive error handling."""
    try:
        result = tool_func(**args)
        return json.dumps({"status": "success", "result": result})
    except ToolExecutionError as e:
        return json.dumps({
            "status": "error",
            "error_type": e.error_type,
            "message": e.message,
            "suggestion": "Try a different approach or parameters.",
        })
    except TypeError as e:
        return json.dumps({
            "status": "error",
            "error_type": "invalid_arguments",
            "message": f"Invalid arguments for {tool_name}: {str(e)}",
            "suggestion": "Check the parameter types and try again.",
        })
    except Exception as e:
        return json.dumps({
            "status": "error",
            "error_type": "unexpected_error",
            "message": f"Unexpected error in {tool_name}: {str(e)}",
            "traceback": traceback.format_exc(),
        })

"""Agent with self-correction on tool errors."""

def run_agent_with_self_correction(user_message: str, max_iterations: int = 15) -> str:
    messages = [
        {
            "role": "system",
            "content": (
                "You are a helpful assistant with tools. "
                "If a tool call returns an error, analyze the error message "
                "and try a different approach. Common fixes:\n"
                "- If a file is not found, try listing the directory first.\n"
                "- If a search returns no results, rephrase the query.\n"
                "- If a calculation fails, break it into simpler steps.\n"
                "Do not repeat the exact same tool call that failed."
            )
        },
        {"role": "user", "content": user_message}
    ]

    failed_calls = set()  # Track failed tool calls to avoid repetition

    for i in range(max_iterations):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto",
        )

        message = response.choices[0].message
        messages.append(message)

        if not message.tool_calls:
            return message.content

        for tool_call in message.tool_calls:
            call_signature = f"{tool_call.function.name}:{tool_call.function.arguments}"

            if call_signature in failed_calls:
                # Prevent repeating the exact same failed call
                result_str = json.dumps({
                    "status": "error",
                    "message": "This exact call already failed. Please try a different approach."
                })
            else:
                result_str = execute_tool_safely(
                    tool_call.function.name,
                    TOOL_FUNCTIONS.get(tool_call.function.name),
                    json.loads(tool_call.function.arguments)
                )

                result = json.loads(result_str)
                if result.get("status") == "error":
                    failed_calls.add(call_signature)

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result_str,
            })

    return "Agent reached maximum iterations."

"""Example of a sandboxed tool execution environment."""

import os
import tempfile

class SandboxedEnvironment:
    """A restricted execution environment for agent tools."""

    def __init__(self, allowed_dirs: list[str], max_file_size: int = 10_000_000):
        self.allowed_dirs = [os.path.abspath(d) for d in allowed_dirs]
        self.max_file_size = max_file_size
        self.temp_dir = tempfile.mkdtemp(prefix="agent_sandbox_")
        self.allowed_dirs.append(self.temp_dir)

    def validate_path(self, path: str) -> str:
        """Validate that a path is within allowed directories."""
        abs_path = os.path.abspath(path)
        for allowed in self.allowed_dirs:
            if abs_path.startswith(allowed):
                return abs_path
        raise PermissionError(
            f"Access denied: {path} is outside allowed directories. "
            f"Allowed: {self.allowed_dirs}"
        )

    def validate_command(self, command: str) -> bool:
        """Check if a shell command is allowed."""
        BLOCKED_PATTERNS = [
            "rm -rf /", "sudo", "chmod 777", "curl | sh",
            "wget | sh", "> /dev/", "mkfs", "dd if=",
        ]
        for pattern in BLOCKED_PATTERNS:
            if pattern in command:
                raise PermissionError(f"Blocked command pattern: {pattern}")
        return True

"""Permission system for agent tool calls."""

from enum import Enum

class PermissionLevel(Enum):
    AUTO = "auto"            # Agent can execute freely
    NOTIFY = "notify"        # Execute but notify the user
    CONFIRM = "confirm"      # Require user confirmation before executing
    DENY = "deny"            # Never allow

TOOL_PERMISSIONS = {
    "web_search": PermissionLevel.AUTO,
    "calculator": PermissionLevel.AUTO,
    "read_file": PermissionLevel.AUTO,
    "write_file": PermissionLevel.NOTIFY,
    "execute_code": PermissionLevel.CONFIRM,
    "send_email": PermissionLevel.CONFIRM,
    "delete_file": PermissionLevel.CONFIRM,
    "shell_command": PermissionLevel.CONFIRM,
    "make_payment": PermissionLevel.DENY,
}

"""
Complete multi-tool agent implementation.

This agent can:
1. Perform mathematical calculations
2. Search the web (simulated)
3. Read files from a designated directory
"""

import json
import math
import os
from datetime import datetime
from openai import OpenAI

client = OpenAI()

# --- Configuration ---
WORKSPACE_DIR = "/tmp/agent_workspace"
os.makedirs(WORKSPACE_DIR, exist_ok=True)

# Create some sample files for the agent to work with
with open(os.path.join(WORKSPACE_DIR, "sales_data.txt"), "w") as f:
    f.write("Q1 2025: $142,000\nQ2 2025: $168,000\nQ3 2025: $155,000\nQ4 2025: $193,000\n")

with open(os.path.join(WORKSPACE_DIR, "team.txt"), "w") as f:
    f.write("Alice Chen - Engineering Lead\nBob Kumar - Product Manager\nCarla Diaz - Designer\n")


# --- Tool Implementations ---

def calculator(expression: str) -> dict:
    """Safely evaluate a mathematical expression."""
    allowed = {
        "__builtins__": {},
        "abs": abs, "round": round, "min": min, "max": max,
        "sqrt": math.sqrt, "log": math.log, "log10": math.log10,
        "sin": math.sin, "cos": math.cos, "pow": pow,
        "pi": math.pi, "e": math.e,
    }
    try:
        result = eval(expression, allowed)
        return {"expression": expression, "result": result}
    except Exception as e:
        return {"error": f"Cannot evaluate '{expression}': {str(e)}"}


def web_search(query: str, max_results: int = 3) -> dict:
    """Simulated web search (replace with real API in production)."""
    simulated_db = {
        "python": [
            {"title": "Python.org", "url": "https://python.org", "snippet": "The official Python programming language website."},
        ],
        "climate": [
            {"title": "NASA Climate Change", "url": "https://climate.nasa.gov", "snippet": "Vital signs of the planet: global temperature, CO2 levels, sea ice."},
        ],
        "transformer": [
            {"title": "Attention Is All You Need (Vaswani et al., 2017)", "url": "https://arxiv.org/abs/1706.03762", "snippet": "The original Transformer paper that introduced self-attention."},
        ],
    }

    results = []
    query_lower = query.lower()
    for keyword, entries in simulated_db.items():
        if keyword in query_lower:
            results.extend(entries)

    if not results:
        results = [{"title": "No results found", "url": "", "snippet": f"No simulated results for '{query}'."}]

    return {"query": query, "results": results[:max_results]}


def read_file(filepath: str) -> dict:
    """Read a file from the agent's workspace directory."""
    full_path = os.path.join(WORKSPACE_DIR, filepath)
    abs_path = os.path.abspath(full_path)

    if not abs_path.startswith(os.path.abspath(WORKSPACE_DIR)):
        return {"error": "Access denied: path outside workspace directory."}

    try:
        with open(abs_path, 'r') as f:
            content = f.read()
        return {"filepath": filepath, "content": content, "size_bytes": len(content.encode())}
    except FileNotFoundError:
        available = os.listdir(WORKSPACE_DIR)
        return {"error": f"File not found: {filepath}", "available_files": available}
    except Exception as e:
        return {"error": str(e)}


# --- Tool Registry ---
TOOL_FUNCTIONS = {
    "calculator": calculator,
    "web_search": web_search,
    "read_file": read_file,
}

# --- Tool Definitions for the API ---
TOOL_DEFINITIONS = [
    {
        "type": "function",
        "function": {
            "name": "calculator",
            "description": (
                "Evaluate a mathematical expression. Supports basic arithmetic "
                "(+, -, *, /, **), functions (sqrt, log, log10, sin, cos), "
                "and constants (pi, e). Use this for any calculation that "
                "requires precision."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "A Python mathematical expression. Examples: '2**10', 'sqrt(144)', 'log10(1000)'"
                    }
                },
                "required": ["expression"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "web_search",
            "description": (
                "Search the web for information. Use this when you need "
                "current information, facts you are not confident about, "
                "or links to resources. Returns titles, URLs, and snippets."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "The search query. Be specific and focused."
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "Maximum number of results (default: 3)",
                        "default": 3
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": (
                "Read the contents of a file from the workspace directory. "
                "If the file is not found, returns a list of available files. "
                "Only files in the workspace can be accessed."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "filepath": {
                        "type": "string",
                        "description": "Path to the file relative to the workspace directory. Example: 'data.txt'"
                    }
                },
                "required": ["filepath"]
            }
        }
    }
]


# --- The Agent ---
def run_multi_tool_agent(
    user_query: str,
    max_iterations: int = 10,
    verbose: bool = True
) -> str:
    """Run the multi-tool agent."""
    messages = [
        {
            "role": "system",
            "content": (
                "You are a helpful research assistant with access to tools.\n\n"
                "Available tools:\n"
                "- calculator: For precise mathematical calculations\n"
                "- web_search: For finding current information online\n"
                "- read_file: For reading files from the workspace\n\n"
                "Guidelines:\n"
                "1. Use tools when they would improve accuracy or access needed information.\n"
                "2. Do NOT use the calculator for trivial arithmetic (e.g., 2+2).\n"
                "3. Think step by step for complex tasks.\n"
                "4. If a tool call fails, try a different approach.\n"
                "5. Always provide a clear, complete answer to the user's question.\n"
                f"6. Current date: {datetime.now().strftime('%Y-%m-%d')}"
            )
        },
        {"role": "user", "content": user_query}
    ]

    for iteration in range(max_iterations):
        if verbose:
            print(f"\n--- Iteration {iteration + 1} ---")

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=TOOL_DEFINITIONS,
            tool_choice="auto",
        )

        message = response.choices[0].message
        messages.append(message)

        if message.tool_calls:
            for tool_call in message.tool_calls:
                func_name = tool_call.function.name
                func_args = json.loads(tool_call.function.arguments)

                if verbose:
                    print(f"  Tool: {func_name}({json.dumps(func_args)})")

                if func_name in TOOL_FUNCTIONS:
                    try:
                        result = TOOL_FUNCTIONS[func_name](**func_args)
                    except Exception as e:
                        result = {"error": f"Tool execution failed: {str(e)}"}
                else:
                    result = {"error": f"Unknown tool: {func_name}"}

                result_str = json.dumps(result, indent=2)

                if verbose:
                    print(f"  Result: {result_str[:200]}{'...' if len(result_str) > 200 else ''}")

                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result_str,
                })
        else:
            final_response = message.content
            if verbose:
                print(f"\n  Final response generated ({len(final_response)} chars)")
            return final_response

    return "Agent reached maximum iterations without producing a final answer."


# --- Example Usage ---
if __name__ == "__main__":
    # Example 1: Requires calculation
    print("=" * 60)
    print("Query 1: Compound interest calculation")
    result = run_multi_tool_agent(
        "If I invest $50,000 at 6.5% annual compound interest for 25 years, "
        "how much will I have? What if I also add $500 per month?"
    )
    print(f"\n{result}")

    # Example 2: Requires file reading + calculation
    print("\n" + "=" * 60)
    print("Query 2: Analyze sales data from file")
    result = run_multi_tool_agent(
        "Read the sales_data.txt file and calculate the total annual revenue "
        "and the average quarterly revenue."
    )
    print(f"\n{result}")

"""
A simple MCP server that exposes weather tools and configuration resources.

Install: pip install mcp[cli]
Run:     python weather_server.py
         or: mcp dev weather_server.py (for the MCP Inspector UI)
"""

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("weather-server")


@mcp.tool()
async def get_weather(city: str) -> str:
    """Get current weather for a city.

    Args:
        city: Name of the city (e.g., "Madrid", "London", "Tokyo")

    Returns:
        A string describing the current weather conditions.
    """
    weather_db = {
        "Madrid": {"temp": 28, "condition": "Sunny", "humidity": 35},
        "London": {"temp": 14, "condition": "Overcast", "humidity": 78},
        "Tokyo": {"temp": 22, "condition": "Partly cloudy", "humidity": 60},
    }

    data = weather_db.get(city, {"temp": 20, "condition": "Unknown", "humidity": 50})
    return (
        f"Weather in {city}: {data['temp']}°C, {data['condition']}, "
        f"humidity {data['humidity']}%"
    )


@mcp.tool()
async def get_forecast(city: str, days: int = 3) -> str:
    """Get weather forecast for a city.

    Args:
        city: Name of the city
        days: Number of days to forecast (1-7, default 3)
    """
    if days < 1 or days > 7:
        return "Error: days must be between 1 and 7."

    forecasts = []
    for day in range(1, days + 1):
        forecasts.append(f"  Day {day}: {20 + day}°C, {'Sunny' if day % 2 == 0 else 'Cloudy'}")

    return f"Forecast for {city} ({days} days):\n" + "\n".join(forecasts)


@mcp.resource("config://settings")
async def get_settings() -> str:
    """Return server configuration."""
    return (
        "Weather Server Settings:\n"
        "  max_forecast_days: 7\n"
        "  supported_units: celsius, fahrenheit\n"
        "  rate_limit: 100 requests/hour\n"
        "  data_source: OpenWeatherMap API v3.0"
    )


@mcp.resource("data://supported-cities")
async def get_supported_cities() -> str:
    """Return the list of cities with real-time data."""
    cities = ["Madrid", "London", "Tokyo", "New York", "Paris", "Berlin", "Sydney"]
    return "Supported cities:\n" + "\n".join(f"  - {city}" for city in cities)


@mcp.prompt()
async def travel_weather_check(destination: str, travel_date: str) -> str:
    """Generate a prompt to check weather conditions for travel planning."""
    return (
        f"I am planning to travel to {destination} on {travel_date}. "
        f"Please check the current weather and forecast for {destination}, "
        f"and advise me on what to pack and any weather-related concerns."
    )


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

"""
Simplified MCP client that connects to a server and uses its tools.

This demonstrates the core client-side workflow:
1. Connect to an MCP server
2. Discover available tools
3. Let the LLM decide which tools to call
4. Execute tool calls via MCP and feed results back to the LLM
"""

import asyncio
import json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from anthropic import Anthropic


async def run_mcp_agent(user_query: str):
    """Run an agent that uses tools from an MCP server."""

    # --- Step 1: Connect to the MCP server ---
    server_params = StdioServerParameters(
        command="python",
        args=["weather_server.py"],
    )

    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            # Initialize the MCP connection
            await session.initialize()

            # --- Step 2: Discover available tools ---
            tools_result = await session.list_tools()
            print(f"Discovered {len(tools_result.tools)} tools:")
            for tool in tools_result.tools:
                print(f"  - {tool.name}: {tool.description}")

            # Convert MCP tools to Anthropic's tool format
            anthropic_tools = [
                {
                    "name": tool.name,
                    "description": tool.description,
                    "input_schema": tool.inputSchema,
                }
                for tool in tools_result.tools
            ]

            # --- Step 3: Run the agentic loop ---
            client = Anthropic()
            messages = [{"role": "user", "content": user_query}]

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

                # Check if the model wants to call tools
                if response.stop_reason == "tool_use":
                    tool_results = []
                    for block in response.content:
                        if block.type == "tool_use":
                            print(f"\nCalling tool: {block.name}({json.dumps(block.input)})")

                            # --- Step 4: Execute via MCP ---
                            result = await session.call_tool(
                                block.name,
                                arguments=block.input,
                            )

                            print(f"Result: {result.content[0].text}")

                            tool_results.append({
                                "type": "tool_result",
                                "tool_use_id": block.id,
                                "content": result.content[0].text,
                            })

                    # Feed results back to the LLM
                    messages.append({"role": "assistant", "content": response.content})
                    messages.append({"role": "user", "content": tool_results})
                else:
                    # Model is done — extract final text
                    final_text = "".join(
                        block.text for block in response.content if hasattr(block, "text")
                    )
                    print(f"\nAgent response: {final_text}")
                    return final_text


# Run the agent
asyncio.run(run_mcp_agent("What is the weather in Madrid and London?"))

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["weather_server.py"]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_..."
      }
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/student/projects"]
    }
  }
}