update

airs/base.py

@@ -110,3 +110,9 @@ class Base:
         except Exception as e:
             print(f"退出登录出错: {str(e)}")
             return False
+
+    def test(self):
+        """
+        测试方法，返回一个简单的字符串
+        """
+        return "Base class test method called"

apps/airs_tasks.py

@@ -3,6 +3,17 @@ from fastapi import Request
 from mcp.server.fastmcp import FastMCP
 import os # 用于获取环境变量
 from airs.base import Base # 导入 Base 类
+from pydantic import BaseModel, Field # 用于定义数据模型
+from typing import Optional
+
+# 定义 Pydantic 模型
+class TaskCreate(BaseModel):
+    title: str = Field(..., description="任务的标题（必填）")
+    description: Optional[str] = Field(None, description="任务的详细描述（可选）") # 添加描述
+    due_date: Optional[str] = None # 可以是 'YYYY-MM-DD' 格式的字符串
+    status: Optional[str] = None
+    priority: Optional[str] = None
+    parent_task_id: Optional[str] = None # UUID 字符串，可选
 
 # 创建 MCP 应用实例
 class Tasks(Base): # Tasks 类继承自 Base 类
@@ -37,4 +48,37 @@ class Tasks(Base): # Tasks 类继承自 Base 类
         @self.mcp.tool()
         def app01_add_integers(num1: int, num2: int) -> int:
             """Add two integers in App01"""
+            print(self.test()) # 调用 Base 类的 test 方法
             return num1 + num2
+
+        @self.mcp.tool()
+        def add_task(task: TaskCreate) -> dict:
+            """
+            Add a new task to the database.
+    向数据库添加新任务
+
+    Args:
+        task (TaskCreate): 任务数据对象，包含以下属性：
+        - title (str): 任务标题
+        - description (str): 任务描述
+        - due_date (str): 截止日期（YYYY-MM-DD）
+    """
+            try:
+                # 直接使用 Supabase 客户端插入数据
+                data_to_insert = {
+                    'title': task.title,
+                    'description': task.description,
+                    'parent_task_id': task.parent_task_id,
+                    'due_date': task.due_date,
+                    'status': task.status,
+                    'priority': task.priority
+                }
+                response = self.supabase.table('tasks').insert(data_to_insert).execute()
+
+                # 检查 Supabase 响应
+                if response.data:
+                    return {"success": True, "task": response.data[0]}
+                else:
+                    return {"success": False, "error": response.error.message if response.error else "Unknown error"}
+            except Exception as e:
+                return {"success": False, "error": str(e)}

test/test_sb.py

@@ -0,0 +1,64 @@
+from dotenv import load_dotenv # 导入 load_dotenv
+
+# 加载 .env 文件中的环境变量
+load_dotenv()
+
+from supabase import create_client, Client
+import os
+
+# 从环境变量获取 Supabase URL 和 Key
+# 在实际应用中，请确保这些变量已设置
+supabase_url = os.getenv("SUPABASE_URL")
+supabase_key = os.getenv("SUPABASE_KEY")
+
+
+# !!! 请替换为你的 Supabase 项目 URL 和 anon key !!!
+# SUPABASE_URL = "SUPABASE_URL"
+# SUPABASE_KEY = "SUPABASE_KEY"
+
+# 创建 Supabase 客户端
+supabase: Client = create_client(supabase_url, supabase_key)
+
+def test_insert_and_query_data():
+    """
+    测试向 Supabase 插入数据并查询。
+    请确保你的 Supabase 数据库中存在一个名为 'test_table' 的表，
+    并且该表包含 'name' (text) 和 'age' (integer) 字段。
+    """
+    table_name = "test_table" # 替换为你的表名
+
+    print(f"正在向表 '{table_name}' 插入数据...")
+    # 插入数据
+    data_to_insert = {"name": "Test User", "age": 30}
+    response = supabase.table(table_name).insert(data_to_insert).execute()
+
+    if response.data:
+        print(f"数据插入成功: {response.data}")
+    else:
+        print(f"数据插入失败: {response.error}")
+        assert False, f"插入失败: {response.error}"
+
+    print(f"正在从表 '{table_name}' 查询数据...")
+    # 查询数据
+    response = supabase.table(table_name).select("*").eq("name", "Test User").execute()
+
+    if response.data:
+        print(f"数据查询成功: {response.data}")
+        assert len(response.data) > 0, "未查询到数据"
+        assert response.data[0]["name"] == "Test User"
+    else:
+        print(f"数据查询失败: {response.error}")
+        assert False, f"查询失败: {response.error}"
+
+    # 清理数据 (可选，根据测试需求决定是否保留)
+    print(f"正在从表 '{table_name}' 删除数据...")
+    response = supabase.table(table_name).delete().eq("name", "Test User").execute()
+    if response.data:
+        print(f"数据删除成功: {response.data}")
+    else:
+        print(f"数据删除失败: {response.error}")
+
+if __name__ == "__main__":
+    print("开始运行 Supabase 测试...")
+    test_insert_and_query_data()
+    print("Supabase 测试完成。")

test/test_sb_create_table.py

@@ -0,0 +1,113 @@
+from dotenv import load_dotenv # 导入 load_dotenv
+
+# 加载 .env 文件中的环境变量
+load_dotenv()
+
+from supabase import create_client, Client
+import os
+
+# 从环境变量获取 Supabase URL 和 Key
+# 在实际应用中，请确保这些变量已设置
+supabase_url = os.getenv("SUPABASE_URL")
+supabase_key = os.getenv("SUPABASE_KEY")
+
+
+# !!! 请替换为你的 Supabase 项目 URL 和 anon key !!!
+# SUPABASE_URL = "SUPABASE_URL"
+# SUPABASE_KEY = "SUPABASE_KEY"
+
+# 创建 Supabase 客户端
+supabase: Client = create_client(supabase_url, supabase_key)
+
+def create_test_table():
+    """
+    创建 Supabase 中的 'test_table_1' 表。
+    如果表已存在，则不会重复创建。
+    """
+    table_name = "test_table_1"
+    print(f"正在尝试通过 RPC 创建表 '{table_name}'...")
+    
+    # Supabase Python SDK 不直接支持执行任意 DDL SQL。
+    # 最推荐的方式是在 Supabase 控制台手动创建表，或者通过迁移工具。
+    # 如果必须通过代码执行 DDL，通常需要：
+    # 1. 在 Supabase 数据库中创建一个 PostgreSQL 函数来封装 DDL 逻辑。
+    # 2. 然后通过 supabase.rpc() 调用这个函数。
+    
+    # 以下是需要在 Supabase SQL 编辑器中执行的 PostgreSQL 函数定义：
+    print("\n--- 请在 Supabase SQL 编辑器中执行以下 SQL 来创建函数 'create_table_if_not_exists_func' ---")
+    print(f"""
+CREATE OR REPLACE FUNCTION create_table_if_not_exists_func(table_name TEXT)
+RETURNS VOID AS $$
+BEGIN
+    EXECUTE format('
+        CREATE TABLE IF NOT EXISTS %I (
+            id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+            name TEXT,
+            age INT
+        );
+    ', table_name);
+END;
+$$ LANGUAGE plpgsql;
+    """)
+    print("----------------------------------------------------------------------------------\n")
+
+    # 尝试通过 RPC 调用上述 PostgreSQL 函数
+    try:
+        response = supabase.rpc('create_table_if_not_exists_func', {'table_name': table_name}).execute()
+        
+        if response.error:
+            print(f"通过 RPC 调用函数创建表失败: {response.error}")
+            # 如果错误是由于函数不存在，提示用户先创建函数
+            if "function create_table_if_not_exists_func(text) does not exist" in str(response.error):
+                print("错误提示：PostgreSQL 函数 'create_table_if_not_exists_func' 不存在。请先在 Supabase 控制台创建该函数。")
+        else:
+            print(f"通过 RPC 调用函数创建表成功: {response.data}")
+            
+    except Exception as e:
+        print(f"调用 RPC 函数时发生意外错误: {e}")
+
+
+def test_insert_and_query_data():
+    """
+    测试向 Supabase 插入数据并查询。
+    请确保你的 Supabase 数据库中存在一个名为 'test_table_1' 的表，
+    并且该表包含 'name' (text) 和 'age' (integer) 字段。
+    """
+    table_name = "test_table_1" # 替换为你的表名
+
+    print(f"正在向表 '{table_name}' 插入数据...")
+    # 插入数据
+    data_to_insert = {"name": "Test User", "age": 30}
+    response = supabase.table(table_name).insert(data_to_insert).execute()
+
+    if response.data:
+        print(f"数据插入成功: {response.data}")
+    else:
+        print(f"数据插入失败: {response.error}")
+        assert False, f"插入失败: {response.error}"
+
+    print(f"正在从表 '{table_name}' 查询数据...")
+    # 查询数据
+    response = supabase.table(table_name).select("*").eq("name", "Test User").execute()
+
+    if response.data:
+        print(f"数据查询成功: {response.data}")
+        assert len(response.data) > 0, "未查询到数据"
+        assert response.data[0]["name"] == "Test User"
+    else:
+        print(f"数据查询失败: {response.error}")
+        assert False, f"查询失败: {response.error}"
+
+    # 清理数据 (可选，根据测试需求决定是否保留)
+    print(f"正在从表 '{table_name}' 删除数据...")
+    response = supabase.table(table_name).delete().eq("name", "Test User").execute()
+    if response.data:
+        print(f"数据删除成功: {response.data}")
+    else:
+        print(f"数据删除失败: {response.error}")
+
+if __name__ == "__main__":
+    print("开始运行 Supabase 测试...")
+    create_test_table() # 在运行测试之前尝试创建表
+    test_insert_and_query_data() # 取消注释，以便在创建表后运行数据操作测试
+    print("Supabase 测试完成。")