tools.py

from typing import Dict, Any, List, TypedDict, Optional
from langgraph.graph import Graph, StateGraph
from langgraph.prebuilt import ToolNode
from duckduckgo_search import DDGS

# --- Calculator Tool Types ---

class CalculatorInput(TypedDict):
    operation: str
    numbers: List[float]

class CalculatorOutput(TypedDict):
    result: float
    operation: str

class CalculatorState(TypedDict):
    input: CalculatorInput
    output: Optional[CalculatorOutput]

# --- Search Tool Types ---

class SearchInput(TypedDict):
    query: str
    max_results: int

class SearchResult(TypedDict):
    title: str
    link: str
    snippet: str

class SearchOutput(TypedDict):
    results: List[SearchResult]
    query: str

class SearchState(TypedDict):
    input: SearchInput
    output: Optional[SearchOutput]

# --- Calculator Tool Implementation ---

def create_calculator_tool() -> Graph:
    """Creates a calculator tool using LangGraph that can perform basic arithmetic operations."""
    print("Creating calculator tool")

    def calculator_function(state: CalculatorState) -> Dict[str, Any]:
        print("Calculator function called")
        input_data = state["input"]
        
        if len(input_data["numbers"]) < 2:
            raise ValueError("At least two numbers are required for calculation")

        result = input_data["numbers"][0]
        for num in input_data["numbers"][1:]:
            if input_data["operation"] == "add":
                result += num
            elif input_data["operation"] == "subtract":
                result -= num
            elif input_data["operation"] == "multiply":
                result *= num
            elif input_data["operation"] == "divide":
                if num == 0:
                    raise ValueError("Cannot divide by zero")
                result /= num
            else:
                raise ValueError(f"Unsupported operation: {input_data['operation']}")

        return {
            "output": {
                "result": result,
                "operation": input_data["operation"]
            }
        }
    print("Calculator function defined")
    workflow = StateGraph(state_schema=CalculatorState)
    print("Calculator workflow created")
    workflow.add_node("calculator", ToolNode(calculator_function))
    print("Calculator tool node added")
    workflow.set_entry_point("calculator")
    print("Calculator workflow set entry point")
    workflow.set_finish_point("calculator")
    print("Calculator workflow set finish point")
    print("Calculator workflow created and compiled")
    return workflow.compile()

# --- Search Tool Implementation ---

def create_search_tool() -> Graph:
    """Creates a search tool using DuckDuckGo that can search for information online."""
    print("Creating search tool")

    def search_function(state: SearchState) -> Dict[str, Any]:
        print("Search function called")
        query = state["input"]["query"]
        max_results = state["input"].get("max_results", 3)

        with DDGS() as ddgs:
            raw_results = list(ddgs.text(query, max_results=max_results))

        results = []
        for r in raw_results:
            try:
                results.append({
                    "title": r.get("title", ""),
                    "link": r.get("href", r.get("link", "")),
                    "snippet": r.get("body", r.get("snippet", ""))
                })
            except Exception as e:
                print("Skipping malformed search result:", r, "Error:", e)

        return {
            "output": {
                "results": results,
                "query": query
            }
        }

    workflow = StateGraph(state_schema=SearchState)
    workflow.add_node("search", ToolNode(search_function))
    workflow.set_entry_point("search")
    workflow.set_finish_point("search")
    print("Search workflow created and compiled")
    return workflow.compile()