tools_registry.py

# Import tools from local extra_tools.py
try:
    from extra_tools import TOOL_REGISTRY as EXTRA_TOOLS
except ImportError:
    EXTRA_TOOLS = {}

# Google Sheets reading tool
def read_google_sheet(url, gid=None):
    """
    Reads the first worksheet of a public Google Sheet and returns its content as a table.
    """
    print("Reading Google Sheet from URL:", url)
    import gspread
    import pandas as pd
    try:
        def extract_sheet_id(url, gid=None):
            import re
            match = re.search(r'/d/([\w-]+)', url)
            return match.group(1) if match else None
        sheet_id = extract_sheet_id(url)
        if gid is None:
            gid = "0"
        csv_url = f"https://docs.google.com/spreadsheets/d/{sheet_id}/export?format=csv&gid={gid}"
        df = pd.read_csv(csv_url)
        df.head()
        return df.to_string(index=False)
    except Exception as e:
        return f"Failed to read Google Sheet: {e}"

# --- Task editing and deletion tools ---

def edit_task(task_id, description=None, deadline=None, type_=None, status=None):
    """
    Edit a task's fields by its unique id. Only provided fields are updated.
    """
    from app_merlin_ai_coach import session_memory  # Import moved inside function
    for t in session_memory.tasks:
        if t.get("id") == task_id:
            if description is not None:
                t["description"] = description
            if deadline is not None:
                t["deadline"] = deadline
            if type_ is not None:
                t["type"] = type_
            if status is not None:
                t["status"] = status
            return f"Task {task_id} updated."
    return f"Task {task_id} not found."

def delete_task(task_id):
    """
    Delete a task by its unique id.
    """
    from app_merlin_ai_coach import session_memory  # Import moved inside function
    before = len(session_memory.tasks)
    session_memory.tasks = [t for t in session_memory.tasks if t.get("id") != task_id]
    after = len(session_memory.tasks)
    if before == after:
        return f"Task {task_id} not found."
    return f"Task {task_id} deleted."

TOOL_REGISTRY = {
    **EXTRA_TOOLS,
    "read_google_sheet": {
        "description": "Read a public Google Sheet and return its content as a table. Usage: read_google_sheet(url, gid (Optional))",
        "function": read_google_sheet,
    },
    "edit_task": {
        "description": "Edit a task by id. Usage: edit_task(task_id, description=..., deadline=..., type_=..., status=...). Only provide fields you want to change.",
        "function": edit_task,
    },
    "delete_task": {
        "description": "Delete a task by id. Usage: delete_task(task_id)",
        "function": delete_task,
    },
    # Add more tools here as needed
}

def call_tool(tool_name, *args, **kwargs):
    """
    Calls a registered tool by name.
    """
    tool = TOOL_REGISTRY.get(tool_name)
    if not tool:
        return f"Tool '{tool_name}' not found."
    try:
        return tool["function"](*args, **kwargs)
    except Exception as e:
        return f"Error running tool '{tool_name}': {e}"

def get_tool_descriptions():
    """
    Returns a string describing all available tools for the system prompt.
    """
    descs = []
    # Add system instruction about no nested tool calls
    # descs.append("System instruction: Tool calls cannot be nested. Do not call a tool/function within another tool/function call.")
    for name, tool in TOOL_REGISTRY.items():
        descs.append(f"{name}: {tool['description']}")
    return "\n".join(descs)

def get_tool_functions():
    """
    Returns a list of tool functions for use with LangChain/LangGraph ToolNode.
    """
    return [tool["function"] for tool in TOOL_REGISTRY.values()]