Agent Tools & MCP

User: "What's the weather in New York?"

[System injects available tools into context]
Available tools:
- get_weather(city, unit)
- search_web(query)
- get_time(timezone)

[LLM decides which tool to call]
LLM Output:
{
  "type": "tool_use",
  "id": "call_1",
  "name": "get_weather",
  "input": {
    "city": "New York",
    "unit": "fahrenheit"
  }
}

[Host executes tool]
Tool Result: {
  "city": "New York",
  "temp": 72,
  "condition": "Partly Cloudy",
  "humidity": 65
}

[System adds result to conversation]
User Message: "What's the weather in New York?"
Assistant: [tool call above]
Tool Result: {temp: 72, ...}

[LLM continues reasoning with new context]
LLM Output (text): "It's currently 72°F and partly cloudy in New York."

1. Call weather_api("NYC")
2. Get result
3. Call weather_api("LA")
4. Get result
5. Compare temperatures

LLM can instead do this in one turn:
{
  "type": "tool_use",
  "tools": [
    {
      "name": "get_weather",
      "input": {"city": "New York"}
    },
    {
      "name": "get_weather",
      "input": {"city": "Los Angeles"}
    }
  ]
}

WEAK description:
"Searches for documents"

STRONG description:
"Search a document database for relevant passages. Use this tool when the user asks a question that requires information from stored documents. Searches by keyword and returns up to 10 results ranked by relevance. Does NOT perform calculations—use calculator_tool for math. Returns empty list if no documents match."

{
  "name": "send_email",
  "description": "Send an email to a recipient",
  "input_schema": {
    "type": "object",
    "properties": {
      "to": {
        "type": "string",
        "description": "Recipient email address (e.g., user@example.com)"
      },
      "subject": {
        "type": "string",
        "description": "Email subject line (max 100 characters)"
      },
      "body": {
        "type": "string",
        "description": "Email body; supports plain text and markdown"
      },
      "cc": {
        "type": "array",
        "items": {"type": "string"},
        "description": "Optional CC recipients (array of emails)"
      }
    },
    "required": ["to", "subject", "body"]
  }
}

def calculate_expression(expression: str) -> dict:
    """Evaluate a mathematical expression safely."""
    try:
        result = eval(expression)
        return {"success": True, "result": result}
    except Exception as e:
        return {"success": False, "error": str(e)}

async def upload_large_file(file_path: str) -> dict:
    """Start file upload and return job ID."""
    job_id = start_upload(file_path)
    return {"job_id": job_id, "status": "uploading"}

# LLM can then poll:
def check_upload_status(job_id: str) -> dict:
    """Check status of upload job."""
    status = get_upload_status(job_id)
    return {"job_id": job_id, "status": status, "progress": ...}

def robust_tool(param: str) -> dict:
    """Tool with comprehensive error handling."""
    # Validate input
    if not param:
        return {"error": "Parameter is empty", "code": "INVALID_INPUT"}

    try:
        # Main logic
        result = perform_operation(param)
        return {"success": True, "data": result}

    except TimeoutError:
        return {"error": "Operation timed out", "code": "TIMEOUT"}
    except PermissionError:
        return {"error": "Access denied", "code": "PERMISSION_DENIED"}
    except Exception as e:
        return {"error": str(e), "code": "INTERNAL_ERROR"}

import signal

def timeout_handler(signum, frame):
    raise TimeoutError("Tool execution exceeded 30 seconds")

def run_with_timeout(tool_func, args, timeout_seconds=30):
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(timeout_seconds)
    try:
        result = tool_func(*args)
        signal.alarm(0)  # Cancel alarm
        return result
    except TimeoutError:
        return {"error": "Execution timeout", "code": "TIMEOUT"}

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": {...},
    "clientInfo": {
      "name": "claude-host",
      "version": "1.0"
    }
  }
}

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "protocolVersion": "2024-11-05",
    "serverInfo": {
      "name": "github-mcp-server",
      "version": "1.0.0"
    },
    "capabilities": {
      "tools": {
        "listChanged": false
      },
      "resources": {
        "subscribe": true
      },
      "prompts": {}
    }
  }
}

import mcp

class MyMCPServer(mcp.Server):
    def __init__(self):
        super().__init__("my-server")

    async def initialize(self):
        """Called when client connects."""
        return {
            "name": "my-server",
            "version": "1.0.0"
        }

    async def list_tools(self):
        """Return list of available tools with schemas."""
        return [
            {
                "name": "add_numbers",
                "description": "Add two numbers",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "a": {"type": "number"},
                        "b": {"type": "number"}
                    },
                    "required": ["a", "b"]
                }
            }
        ]

    async def call_tool(self, name: str, arguments: dict):
        """Execute a tool."""
        if name == "add_numbers":
            return arguments["a"] + arguments["b"]
        else:
            raise ValueError(f"Unknown tool: {name}")

if __name__ == "__main__":
    server = MyMCPServer()
    server.run()  # Runs over stdio

@server.list_tools()
async def list_tools():
    return [
        {
            "name": "calculate",
            "description": "Evaluate math expression",
            "inputSchema": {...}
        },
        {
            "name": "search_documents",
            "description": "Search document store",
            "inputSchema": {...}
        }
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "calculate":
        return eval(arguments["expression"])
    elif name == "search_documents":
        return search_db(arguments["query"])
    else:
        raise ValueError(f"Unknown tool: {name}")

# List and read resources
@server.list_resources()
async def list_resources():
    return [{"uri": "file:///config.json", "name": "Config"}]

@server.read_resource()
async def read_resource(uri: str):
    return open(uri).read()

# Expose prompt templates
@server.list_prompts()
async def list_prompts():
    return [{"name": "code_review", "description": "Review code"}]

@server.get_prompt()
async def get_prompt(name: str, arguments: dict):
    if name == "code_review":
        return f"Review: {arguments['code']}"

async def call_tool(name: str, arguments: dict):
    if name == "delete_file":
        path = arguments["path"]
        print(f"Agent wants to delete: {path}")
        if input("Approve? (yes/no): ").lower() != "yes":
            return {"error": "Denied"}
        os.remove(path)
        return {"success": True}
    elif name == "search":  # Safe
        return search(arguments["query"])

import subprocess

def execute_code_safely(code: str, timeout: int = 5):
    """Execute code in sandbox. Returns stdout."""
    try:
        result = subprocess.run(
            ["python", "-c", code],
            capture_output=True,
            text=True,
            timeout=timeout,
            cwd="/tmp/sandbox"  # Isolated dir
        )
        return result.stdout
    except subprocess.TimeoutExpired:
        return "ERROR: Timeout"
    except Exception as e:
        return f"ERROR: {e}"

def validate_file_operation(path: str):
    """Ensure path is safe."""
    abs_path = os.path.abspath(path)
    if not abs_path.startswith("/safe/directory"):
        raise ValueError(f"Path outside safe zone: {abs_path}")
    if not os.path.exists(abs_path):
        raise FileNotFoundError(f"Path not found: {abs_path}")
    return abs_path

import json
from datetime import datetime

logger = logging.getLogger("tool_audit")

async def call_tool_with_audit(name: str, arguments: dict, user: str):
    """Call tool and log everything."""
    log_entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "tool": name,
        "user": user,
        "arguments": arguments
    }
    try:
        result = await execute_tool(name, arguments)
        log_entry["status"] = "success"
        log_entry["result_summary"] = str(result)[:100]
    except Exception as e:
        log_entry["status"] = "error"
        log_entry["error"] = str(e)
        raise
    finally:
        logger.info(json.dumps(log_entry))

from collections import defaultdict
from time import time

class RateLimiter:
    def __init__(self, max_calls: int = 100, window: int = 60):
        self.max_calls = max_calls
        self.window = window
        self.calls = defaultdict(list)

    def allow_call(self, tool_name: str) -> bool:
        """Check if tool call is allowed."""
        now = time()
        self.calls[tool_name] = [
            t for t in self.calls[tool_name]
            if now - t < self.window
        ]
        if len(self.calls[tool_name]) >= self.max_calls:
            return False
        self.calls[tool_name].append(now)
        return True

def get_relevant_tools(query: str) -> list:
    """Filter tools by query type."""
    if "weather" in query.lower():
        return [weather_tool, location_tool]
    elif "calculate" in query.lower():
        return [calculator_tool, converter_tool]
    elif "file" in query.lower():
        return [read_file, list_files]
    else:
        return [search_tool]

# LLM only sees 2-4 tools, not 20
relevant = get_relevant_tools(user_query)
llm_response = llm.call(
    messages=user_messages,
    tools=relevant
)

tools = [
    {
        "name": "database_query",
        "description": "Execute SQL",
        "input_schema": {
            "operation": {"enum": ["select","insert","update"]}
        }
    }
]

async def database_query(query: str, operation: str):
    if operation == "select":
        return execute_select(query)
    elif operation == "insert":
        return execute_insert(query)
    elif operation == "update":
        return execute_update(query)

async def robust_search(query: str):
    try:
        return search_primary(query)  # Primary
    except TimeoutError:
        return search_cache(query)  # Fallback
    except Exception:
        return {"error": "Search unavailable"}

import asyncio

async def get_user_profile(user_id: str):
    profile, messages, settings = await asyncio.gather(
        fetch_profile(user_id),
        fetch_messages(user_id),
        fetch_settings(user_id)
    )
    return {
        "profile": profile,
        "messages": messages,
        "settings": settings
    }

User: "What's the weather in NYC?"

[TRACE]
Reasoning: "User asked about weather"
Tool: get_weather
Arguments: {"city": "New York", "unit": "fahrenheit"}
Result: {"temp": 72, "condition": "Partly cloudy"}
Response: "It's 72°F and partly cloudy in NYC"

Tool not called?
→ Check tool description (too vague?)
→ Check if tool shown to LLM?

Wrong arguments?
→ Check schema clarity
→ Check examples in description

Result not used?
→ Is result JSON or plain text?
→ Check error indicator

LLM refusing tool?
→ Security guard triggered?
→ Tool name seems malicious?

import unittest

class TestCalculatorTool(unittest.TestCase):
    def test_add(self):
        result = calculate_tool("2 + 2")
        self.assertEqual(result, 4)

    def test_syntax_error(self):
        result = calculate_tool("2 +")
        self.assertEqual(result["error"], "Syntax error")

    def test_timeout(self):
        result = calculate_tool("while True: pass")
        self.assertEqual(result["error"], "Timeout")

class TestAgentWorkflow(unittest.TestCase):
    def test_weather_query(self):
        """Test agent uses weather tool."""
        agent = Agent(tools=[weather_tool, search_tool])
        response = agent.run("What's the weather in NYC?")
        self.assertIn("tool_call", response["trace"])
        self.assertIn("72", response["text"])

import json
from typing import Any, Dict, Optional

class Tool:
    """Base class for tools."""
    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description

    def execute(self, **kwargs) -> Dict[str, Any]:
        """Override in subclasses."""
        raise NotImplementedError

    def schema(self) -> Dict[str, Any]:
        """Return JSON schema."""
        return {
            "name": self.name,
            "description": self.description,
            "input_schema": self._input_schema()
        }

    def _input_schema(self) -> Dict[str, Any]:
        """Return parameter schema."""
        raise NotImplementedError

class CalculatorTool(Tool):
    def __init__(self):
        super().__init__(
            "calculate",
            "Evaluate a math expression safely"
        )

    def execute(self, expression: str) -> Dict[str, Any]:
        try:
            result = eval(expression)
            return {"success": True, "result": result}
        except Exception as e:
            return {"success": False, "error": str(e)}

    def _input_schema(self):
        return {
            "type": "object",
            "properties": {
                "expression": {"type": "string"}
            },
            "required": ["expression"]
        }

class SearchTool(Tool):
    def __init__(self, documents: list):
        super().__init__(
            "search_documents",
            "Search documents by keyword"
        )
        self.documents = documents

    def execute(self, query: str) -> Dict[str, Any]:
        query_lower = query.lower()
        results = [
            doc for doc in self.documents
            if query_lower in doc.lower()
        ]
        return {
            "success": True,
            "results": results[:5],
            "count": len(results)
        }

    def _input_schema(self):
        return {
            "type": "object",
            "properties": {
                "query": {"type": "string"}
            },
            "required": ["query"]
        }

class WeatherTool(Tool):
    def __init__(self):
        super().__init__(
            "get_weather",
            "Get weather for a city"
        )
        self.data = {
            "new york": {"temp": 72, "condition": "Cloudy"},
            "london": {"temp": 59, "condition": "Rainy"},
            "tokyo": {"temp": 81, "condition": "Sunny"}
        }

    def execute(self, city: str) -> Dict[str, Any]:
        city_lower = city.lower()
        if city_lower in self.data:
            return {
                "success": True,
                "city": city,
                "data": self.data[city_lower]
            }
        return {"success": False, "error": f"City not found"}

    def _input_schema(self):
        return {
            "type": "object",
            "properties": {
                "city": {"type": "string"}
            },
            "required": ["city"]
        }

class ToolRegistry:
    """Register and discover tools."""
    def __init__(self):
        self.tools = {}

    def register(self, tool: Tool) -> None:
        """Register a tool."""
        self.tools[tool.name] = tool

    def get_tool(self, name: str) -> Optional[Tool]:
        """Get tool by name."""
        return self.tools.get(name)

    def list_tools(self) -> list:
        """Return all tool schemas."""
        return [tool.schema() for tool in self.tools.values()]

    def call_tool(self, name: str, arguments: Dict) -> Dict:
        """Execute a tool."""
        tool = self.get_tool(name)
        if not tool:
            return {"error": f"Unknown tool: {name}"}
        return tool.execute(**arguments)

import re

class MockLLM:
    """Simulate LLM decision-making."""
    def __init__(self, registry: ToolRegistry):
        self.registry = registry

    def call(self, user_message: str) -> Dict[str, Any]:
        """Decide which tool to call."""
        msg = user_message.lower()

        if any(w in msg for w in ["calculate", "math", "2+2"]):
            match = re.search(r'\b(\d+[\+\-\*/]\d+)\b', user_message)
            expr = match.group(1) if match else "2+2"
            return {
                "success": True,
                "tool_selected": "calculate",
                "arguments": {"expression": expr}
            }

        elif any(w in msg for w in ["weather", "temperature"]):
            city = "new york"
            if "london" in msg:
                city = "london"
            elif "tokyo" in msg:
                city = "tokyo"
            return {
                "success": True,
                "tool_selected": "get_weather",
                "arguments": {"city": city}
            }

        elif any(w in msg for w in ["search", "find"]):
            return {
                "success": True,
                "tool_selected": "search_documents",
                "arguments": {"query": user_message}
            }

        return {
            "success": False,
            "response": "I'm not sure how to help."
        }

class Agent:
    """Reason → Act → Observe loop."""
    def __init__(self, registry: ToolRegistry):
        self.registry = registry
        self.llm = MockLLM(registry)
        self.trace = []

    def run(self, user_message: str) -> Dict[str, Any]:
        """Run agent."""
        self.trace = []

        # Reason
        reasoning = self.llm.call(user_message)
        self.trace.append({"step": "reason", "result": reasoning})

        if not reasoning["success"]:
            return {
                "user_message": user_message,
                "response": reasoning["response"],
                "trace": self.trace
            }

        # Act
        tool_name = reasoning["tool_selected"]
        arguments = reasoning["arguments"]
        tool_result = self.registry.call_tool(tool_name, arguments)
        self.trace.append({
            "step": "act",
            "tool": tool_name,
            "arguments": arguments,
            "result": tool_result
        })

        # Observe
        response = self._generate_response(tool_name, tool_result)
        self.trace.append({"step": "observe", "response": response})

        return {
            "user_message": user_message,
            "response": response,
            "trace": self.trace
        }

    def _generate_response(self, tool: str, result: Dict) -> str:
        """Generate response from tool result."""
        if not result.get("success", True):
            return f"Error: {result.get('error')}"

        if tool == "calculate":
            return f"The answer is {result['result']}"
        elif tool == "get_weather":
            data = result.get("data", {})
            city = result.get("city", "the city")
            return f"In {city}: {data.get('temp', 'N/A')}°F, {data.get('condition')}"
        elif tool == "search_documents":
            count = result.get("count", 0)
            return f"Found {count} matching documents"
        else:
            return "Done"

# Setup
registry = ToolRegistry()
registry.register(CalculatorTool())
registry.register(SearchTool([
    "Claude is an AI assistant",
    "Python is a programming language",
    "Tools extend LLM capabilities"
]))
registry.register(WeatherTool())
agent = Agent(registry)

# Query 1: Math
r = agent.run("What is 10 * 5?")
print(f"User: {r['user_message']}")
print(f"Agent: {r['response']}\n")

# Query 2: Weather
r = agent.run("Tell me the weather in Tokyo")
print(f"User: {r['user_message']}")
print(f"Agent: {r['response']}\n")

# Query 3: Search
r = agent.run("Search for Python in documents")
print(f"User: {r['user_message']}")
print(f"Agent: {r['response']}\n")

# Full trace
print("=== Trace ===")
for step in r["trace"]:
    print(json.dumps(step, indent=2))