Spaces:
Runtime error
Runtime error
| # 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()] | |