Supabase MCP 代理

使用 Supabase MCP 服务器 创建一个代理，该代理可以创建项目、数据库 schema、边缘函数等：

"""🔑 Supabase MCP Agent - 展示 Supabase MCP 的功能

此示例演示如何使用 Supabase MCP 服务器来创建项目、数据库 schema、边缘函数等。

设置：
1. 安装 Python 依赖：`pip install agno mcp-sdk`
2. 创建一个 Supabase 访问令牌：https://supabase.com/dashboard/account/tokens 并将其设置为 SUPABASE_ACCESS_TOKEN 环境变量。
"""

import asyncio
import os
from textwrap import dedent

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.mcp import MCPTools
from agno.tools.reasoning import ReasoningTools
from agno.utils.log import log_error, log_exception, log_info


async def run_agent(task: str) -> None:
    token = os.getenv("SUPABASE_ACCESS_TOKEN")
    if not token:
        log_error("未设置 SUPABASE_ACCESS_TOKEN 环境变量。")
        return

    npx_cmd = "npx.cmd" if os.name == "nt" else "npx"

    try:
        async with MCPTools(
            f"{npx_cmd} -y @supabase/mcp-server-supabase@latest --access-token={token}"
        ) as mcp:
            instructions = dedent(f"""
                你是一位专业的 Supabase MCP 架构师。根据项目描述：
                {task}

                自动执行以下步骤：
                1. 根据项目描述规划整个数据库 schema。
                2. 调用 `list_organizations` 并选择响应中的第一个组织。
                3. 使用 `get_cost(type='project')` 估算项目创建成本，并在你的回复中提及成本。
                4. 使用 `create_project` 创建一个新的 Supabase 项目，并传入确认的成本 ID。
                5. 使用 `get_project` 轮询项目状态，直到状态为 `ACTIVE_HEALTHY`。
                6. 分析项目需求，并提出一个完整的、标准化的 SQL schema（表、列、数据类型、索引、约束、触发器和函数）作为 DDL 语句。
                7. 使用 `apply_migration` 应用 schema，并将迁移命名为 `initial_schema`。
                8. 通过 `list_tables` 和 `list_extensions` 验证已部署的 schema。
                8. 使用 `deploy_edge_function` 部署一个简单的健康检查边缘函数。
                9. 检索并打印项目 URL (`get_project_url`) 和匿名密钥 (`get_anon_key`)。
            """)
            agent = Agent(
                model=OpenAIChat(id="o4-mini"),
                instructions=instructions,
                tools=[mcp, ReasoningTools(add_instructions=True)],
                markdown=True,
            )

            log_info(f"正在为以下项目运行 Supabase 项目代理：{task}")
            await agent.aprint_response(
                message=task,
                stream=True,
                stream_intermediate_steps=True,
                show_full_reasoning=True,
            )
    except Exception as e:
        log_exception(f"发生意外错误：{e}")


if __name__ == "__main__":
    demo_description = (
        "开发一个基于云的 SaaS 平台，具备 AI 驱动的任务建议、日历同步、预测性优先级排序、团队协作和项目分析功能。"
    )
    asyncio.run(run_agent(demo_description))


# 示例提示供尝试：
"""
一个 SaaS 工具，帮助企业使用 AI 自动化文档处理。用户可以上传发票、合同或 PDF，并获得结构化数据、智能摘要以及用于合规性或异常情况的红色标记警报。非常适合法律团队、会计师和企业后台办公室。

一个由 AI 增强的 SaaS 平台，用于简化招聘流程。功能包括使用 NLP 进行自动候选人筛选、AI 面试安排、工作描述中的偏见检测以及流水线分析。专为快速发展的初创公司和中型 HR 团队设计。

一个内部 SaaS 工具，供 HR 部门监控员工的幸福感。结合每周的情绪签到、匿名反馈和 AI 驱动的倦怠检测模型。与 Slack 和 HR 系统集成，以支持更健康的工作场所文化。
"""