M06: Multi-Tool Orchestration

# pip install anthropic>=0.30.0
import anthropic
import json
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY env var

tools = [
    {
        "name": "web_search",
        "description": (
            "Search the web for current information. Returns top 3 "
            "results with title, URL, and snippet. Use for recent "
            "events, factual questions, or general research."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query"}
            },
            "required": ["query"]
        }
    },
    {
        "name": "fetch_page",
        "description": (
            "Fetch the full text content of a web page by URL. "
            "Returns page text (max 5000 chars). Use after "
            "web_search to get full content from a result URL."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "url": {"type": "string", "description": "Full URL to fetch"}
            },
            "required": ["url"]
        }
    },
    {
        "name": "summarize_text",
        "description": (
            "Summarize long text into key points (3-5 bullets). "
            "Use after fetch_page to condense page content."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "text": {"type": "string", "description": "Text to summarize"},
                "max_points": {
                    "type": "integer",
                    "description": "Max bullet points (default 5)"
                }
            },
            "required": ["text"]
        }
    },
    {
        "name": "format_citation",
        "description": (
            "Format a source as an academic citation. Use after "
            "summaries are ready to create proper references."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "title": {"type": "string", "description": "Article title"},
                "url": {"type": "string", "description": "Source URL"},
                "accessed_date": {"type": "string", "description": "e.g. '2025-01-15'"}
            },
            "required": ["title", "url"]
        }
    },
    {
        "name": "save_to_file",
        "description": "Save content to a local file. Returns file path.",
        "input_schema": {
            "type": "object",
            "properties": {
                "filename": {"type": "string", "description": "Output filename"},
                "content": {"type": "string", "description": "Content to save"}
            },
            "required": ["filename", "content"]
        }
    }
]

# Mock implementations (replace with real APIs in production)
def web_search(query: str) -> dict:
    time.sleep(0.2)  # Simulate latency
    return {"results": [
        {"title": f"Result 1: {query}", "url": "https://example.com/1",
         "snippet": f"Overview of {query}..."},
        {"title": f"Result 2: {query}", "url": "https://example.com/2",
         "snippet": f"Developments in {query}..."},
        {"title": f"Result 3: {query}", "url": "https://broken.example.com/404",
         "snippet": f"Deep dive into {query}..."},
    ]}

def fetch_page(url: str) -> dict:
    time.sleep(0.3)
    if "broken" in url or "404" in url:
        raise ConnectionError(f"404 Not Found: {url}")
    return {"content": f"Full page content from {url}. " * 20}

def summarize_text(text: str, max_points: int = 5) -> dict:
    return {"summary": [f"Key point {i+1}" for i in range(min(max_points, 5))]}

def format_citation(title: str, url: str, accessed_date: str = None) -> dict:
    date = accessed_date or "2025-01-15"
    return {"citation": f'"{title}." Available at: {url}. Accessed: {date}.'}

def save_to_file(filename: str, content: str) -> dict:
    return {"status": "saved", "path": f"/output/{filename}", "bytes": len(content)}

# Dispatcher with per-tool error handling
tool_functions = {
    "web_search": web_search, "fetch_page": fetch_page,
    "summarize_text": summarize_text, "format_citation": format_citation,
    "save_to_file": save_to_file,
}

def execute_tool(name: str, inputs: dict) -> tuple[str, bool]:
    """Execute a tool, returning (result_json, is_error)."""
    func = tool_functions.get(name)
    if not func:
        return json.dumps({"error": f"Unknown tool: {name}"}), True
    try:
        result = func(**inputs)
        return json.dumps(result), False
    except Exception as e:
        return json.dumps({"error": str(e)}), True

def run_research_agent(question: str, available_tools=None) -> str:
    """Run the agentic loop with parallel tool execution."""
    active_tools = available_tools or tools
    messages = [{"role": "user", "content": question}]
    max_iterations = 10  # Safety limit

    for iteration in range(max_iterations):
        try:
            response = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=4096,
                system="You are a research assistant. Search multiple "
                       "sources in parallel when possible.",
                tools=active_tools,
                messages=messages,
            )
        except anthropic.APIError as e:
            return f"API error: {e.status_code} - {e.message}"

        # Collect tool_use blocks
        tool_uses = [b for b in response.content if b.type == "tool_use"]

        if response.stop_reason == "end_turn" or not tool_uses:
            # Claude is done — extract final text
            return "\n".join(
                b.text for b in response.content if b.type == "text"
            )

        # Execute tools — parallel when multiple requested
        if len(tool_uses) > 1:
            # PARALLEL: use ThreadPoolExecutor
            tool_results = []
            with ThreadPoolExecutor(max_workers=len(tool_uses)) as pool:
                futures = {
                    pool.submit(execute_tool, tu.name, tu.input): tu.id
                    for tu in tool_uses
                }
                for future in as_completed(futures):
                    tid = futures[future]
                    result_json, is_err = future.result()
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": tid,
                        "content": result_json,
                        **({"is_error": True} if is_err else {}),
                    })
        else:
            # SEQUENTIAL: single tool
            tu = tool_uses[0]
            result_json, is_err = execute_tool(tu.name, tu.input)
            tool_results = [{
                "type": "tool_result",
                "tool_use_id": tu.id,
                "content": result_json,
                **({"is_error": True} if is_err else {}),
            }]

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

    return "Max iterations reached."

# Run it
answer = run_research_agent(
    "Research the latest developments in AI agents. "
    "Search multiple sources and summarize findings."
)
print(answer)

class ToolRegistry:
    """Manages tools and filters them by context."""

    def __init__(self):
        self._tools: dict[str, dict] = {}
        self._tags: dict[str, set[str]] = {}

    def register(self, tool: dict, tags: list[str] = None):
        name = tool["name"]
        self._tools[name] = tool
        self._tags[name] = set(tags or [])

    def unregister(self, name: str):
        self._tools.pop(name, None)
        self._tags.pop(name, None)

    def get_tools_for_context(
        self, tags: list[str] = None, names: list[str] = None
    ) -> list[dict]:
        if names:
            return [self._tools[n] for n in names if n in self._tools]
        if tags:
            tag_set = set(tags)
            return [
                self._tools[n] for n, t in self._tags.items()
                if t & tag_set
            ]
        return list(self._tools.values())

# Usage
registry = ToolRegistry()
registry.register(tools[0], tags=["research", "search"])
registry.register(tools[1], tags=["research", "fetch"])
registry.register(tools[2], tags=["research", "analysis"])
registry.register(tools[3], tags=["citation"])
registry.register(tools[4], tags=["output"])

# Phase 1: only research tools
research_tools = registry.get_tools_for_context(tags=["research"])
# => [web_search, fetch_page, summarize_text]

# Phase 2: add citation tools after summaries are ready
cite_tools = registry.get_tools_for_context(
    names=["format_citation", "save_to_file"]
)

mkdir multi-tool-lab && cd multi-tool-lab
python -m venv venv && source venv/bin/activate   # Windows: venv\Scripts\activate
pip install anthropic>=0.30.0
export ANTHROPIC_API_KEY=your-key-here             # Windows: set ANTHROPIC_API_KEY=your-key-here

import anthropic
import json
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY

# ── Tool Schemas (what Claude sees) ──────────────────────────
tools = [
    {
        "name": "web_search",
        "description": (
            "Search the web for current information. Returns top 3 "
            "results with title, URL, and snippet. Use for recent "
            "events, factual questions, or general research."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query"}
            },
            "required": ["query"]
        }
    },
    {
        "name": "fetch_page",
        "description": (
            "Fetch the full text content of a web page by URL. "
            "Returns page text (max 5000 chars). Use after "
            "web_search to get full content from a result URL."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "url": {"type": "string", "description": "Full URL to fetch"}
            },
            "required": ["url"]
        }
    },
    {
        "name": "summarize_text",
        "description": (
            "Summarize long text into key points (3-5 bullets). "
            "Use after fetch_page to condense page content."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "text": {"type": "string", "description": "Text to summarize"},
                "max_points": {
                    "type": "integer",
                    "description": "Max bullet points (default 5)"
                }
            },
            "required": ["text"]
        }
    },
    {
        "name": "format_citation",
        "description": (
            "Format a source as an academic citation. Use after "
            "summaries are ready to create proper references."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "title": {"type": "string", "description": "Article title"},
                "url": {"type": "string", "description": "Source URL"},
                "accessed_date": {"type": "string", "description": "e.g. '2025-01-15'"}
            },
            "required": ["title", "url"]
        }
    },
    {
        "name": "save_to_file",
        "description": "Save content to a local file. Returns file path.",
        "input_schema": {
            "type": "object",
            "properties": {
                "filename": {"type": "string", "description": "Output filename"},
                "content": {"type": "string", "description": "Content to save"}
            },
            "required": ["filename", "content"]
        }
    }
]

# ── Mock Implementations ─────────────────────────────────────
def web_search(query: str) -> dict:
    time.sleep(0.2)
    return {"results": [
        {"title": f"Result 1: {query}", "url": "https://example.com/1",
         "snippet": f"Overview of {query}..."},
        {"title": f"Result 2: {query}", "url": "https://example.com/2",
         "snippet": f"Developments in {query}..."},
        {"title": f"Result 3: {query}", "url": "https://broken.example.com/404",
         "snippet": f"Deep dive into {query}..."},
    ]}

def fetch_page(url: str) -> dict:
    time.sleep(0.3)
    if "broken" in url or "404" in url:
        raise ConnectionError(f"404 Not Found: {url}")
    return {"content": f"Full page content from {url}. " * 20}

def summarize_text(text: str, max_points: int = 5) -> dict:
    return {"summary": [f"Key point {i+1}" for i in range(min(max_points, 5))]}

def format_citation(title: str, url: str, accessed_date: str = None) -> dict:
    date = accessed_date or "2025-01-15"
    return {"citation": f'"{title}." Available at: {url}. Accessed: {date}.'}

def save_to_file(filename: str, content: str) -> dict:
    return {"status": "saved", "path": f"/output/{filename}", "bytes": len(content)}

# ── Dispatcher with Error Handling ───────────────────────────
tool_functions = {
    "web_search": web_search, "fetch_page": fetch_page,
    "summarize_text": summarize_text, "format_citation": format_citation,
    "save_to_file": save_to_file,
}

def execute_tool(name: str, inputs: dict) -> tuple[str, bool]:
    """Execute a tool, returning (result_json, is_error)."""
    func = tool_functions.get(name)
    if not func:
        return json.dumps({"error": f"Unknown tool: {name}"}), True
    try:
        result = func(**inputs)
        return json.dumps(result), False
    except Exception as e:
        return json.dumps({"error": str(e), "tool": name}), True

# ── ToolRegistry ─────────────────────────────────────────────
class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, dict] = {}
        self._tags: dict[str, set[str]] = {}

    def register(self, tool: dict, tags: list[str] = None):
        self._tools[tool["name"]] = tool
        self._tags[tool["name"]] = set(tags or [])

    def unregister(self, name: str):
        self._tools.pop(name, None)
        self._tags.pop(name, None)

    def get_tools_for_context(self, tags: list[str] = None, names: list[str] = None) -> list[dict]:
        if names:
            return [self._tools[n] for n in names if n in self._tools]
        if tags:
            tag_set = set(tags)
            return [self._tools[n] for n, t in self._tags.items() if t & tag_set]
        return list(self._tools.values())

# Register tools with tags
registry = ToolRegistry()
registry.register(tools[0], tags=["research", "search"])
registry.register(tools[1], tags=["research", "fetch"])
registry.register(tools[2], tags=["research", "analysis"])
registry.register(tools[3], tags=["citation"])
registry.register(tools[4], tags=["output"])

print("✓ Tools, dispatcher, and registry ready.")
print(f"  All tools: {[t['name'] for t in registry.get_tools_for_context()]}")
print(f"  Research only: {[t['name'] for t in registry.get_tools_for_context(tags=['research'])]}")

# ── Agentic Loop with Parallel Execution ─────────────────────
def run_agent(question: str, tool_tags: list[str] = None, verbose: bool = True) -> str:
    """Run the multi-tool agent. Optionally filter tools by tag."""
    if tool_tags:
        active_tools = registry.get_tools_for_context(tags=tool_tags)
    else:
        active_tools = registry.get_tools_for_context()

    if verbose:
        print(f"\n{'='*60}")
        print(f"Question: {question}")
        print(f"Active tools: {[t['name'] for t in active_tools]}")
        print(f"{'='*60}")

    messages = [{"role": "user", "content": question}]
    max_iterations = 10

    for iteration in range(max_iterations):
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            system=(
                "You are a research assistant. When asked to compare or "
                "research multiple topics, search for each one in parallel. "
                "When asked to fetch and summarize a page, do it sequentially."
            ),
            tools=active_tools,
            messages=messages,
        )

        tool_uses = [b for b in response.content if b.type == "tool_use"]

        if response.stop_reason == "end_turn" or not tool_uses:
            final_text = "\n".join(
                b.text for b in response.content if b.type == "text"
            )
            if verbose:
                print(f"\n✓ Agent finished in {iteration + 1} iteration(s)")
            return final_text

        # Show what Claude requested
        if verbose:
            mode = "PARALLEL" if len(tool_uses) > 1 else "SEQUENTIAL"
            print(f"\n  Iteration {iteration + 1} [{mode}]:")
            for tu in tool_uses:
                print(f"    → {tu.name}({json.dumps(tu.input)[:80]}...)")

        # Execute tools — parallel when multiple
        if len(tool_uses) > 1:
            tool_results = []
            with ThreadPoolExecutor(max_workers=len(tool_uses)) as pool:
                futures = {
                    pool.submit(execute_tool, tu.name, tu.input): tu.id
                    for tu in tool_uses
                }
                for future in as_completed(futures):
                    tid = futures[future]
                    result_json, is_err = future.result()
                    if verbose and is_err:
                        print(f"    ✗ Error for {tid}: {result_json[:60]}")
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": tid,
                        "content": result_json,
                        **({"is_error": True} if is_err else {}),
                    })
        else:
            tu = tool_uses[0]
            result_json, is_err = execute_tool(tu.name, tu.input)
            if verbose and is_err:
                print(f"    ✗ Error: {result_json[:60]}")
            tool_results = [{
                "type": "tool_result",
                "tool_use_id": tu.id,
                "content": result_json,
                **({"is_error": True} if is_err else {}),
            }]

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

    return "Max iterations reached."

# ── Test Scenarios ───────────────────────────────────────────
if __name__ == "__main__":
    # Test 1: Parallel search (Claude should call web_search multiple times)
    print("\n" + "▶ TEST 1: PARALLEL SEARCH ".ljust(60, "─"))
    result1 = run_agent(
        "Search for information about these 3 topics: AI agents, "
        "prompt engineering, and tool use patterns.",
        tool_tags=["research"]
    )
    print(f"\nResult preview: {result1[:200]}...")

    # Test 2: Sequential chain (search → fetch → summarize)
    print("\n" + "▶ TEST 2: SEQUENTIAL CHAIN ".ljust(60, "─"))
    result2 = run_agent(
        "Search for 'Claude AI tool use', then fetch the first "
        "result page and summarize its content.",
        tool_tags=["research"]
    )
    print(f"\nResult preview: {result2[:200]}...")

    # Test 3: Error recovery (fetch_page will 404 on broken URL)
    print("\n" + "▶ TEST 3: ERROR RECOVERY ".ljust(60, "─"))
    result3 = run_agent(
        "Fetch and summarize this page: https://broken.example.com/404",
        tool_tags=["research"]
    )
    print(f"\nResult preview: {result3[:200]}...")

    # Test 4: Dynamic tool filtering (citation tools only)
    print("\n" + "▶ TEST 4: DYNAMIC TOOL FILTERING ".ljust(60, "─"))
    result4 = run_agent(
        "Format a citation for an article titled 'Multi-Tool AI Agents' "
        "from https://example.com/agents, accessed today.",
        tool_tags=["citation"]
    )
    print(f"\nResult preview: {result4[:200]}...")