Stripe MCP 代理

使用 Stripe MCP 服务器 来创建一个可以与 Stripe API 交互的代理：

"""💵 Stripe MCP 代理 - 管理您的 Stripe 操作

本示例演示如何创建一个 Agno 代理，通过模型上下文协议 (MCP) 与 Stripe API 进行交互。该代理可以使用自然语言命令创建和管理 Stripe 对象，如客户、产品、价格和支付链接。


设置：
2. 安装 Python 依赖：`pip install agno mcp-sdk`
3. 设置环境变量：export STRIPE_SECRET_KEY=***。

Stripe MCP 文档：https://github.com/stripe/agent-toolkit
"""

import asyncio
import os
from textwrap import dedent

from agno.agent import Agent
from agno.tools.mcp import MCPTools
from agno.utils.log import log_error, log_exception, log_info


async def run_agent(message: str) -> None:
    """
    设置 Stripe MCP 服务器并初始化 Agno 代理
    """
    # 验证 Stripe API 密钥是否可用
    stripe_api_key = os.getenv("STRIPE_SECRET_KEY")
    if not stripe_api_key:
        log_error("未设置 STRIPE_SECRET_KEY 环境变量。")
        return

    enabled_tools = "paymentLinks.create,products.create,prices.create,customers.create,customers.read"

    # 处理不同的操作系统
    npx_command = "npx.cmd" if os.name == "nt" else "npx"

    try:
        # 使用 Stripe 服务器初始化 MCP 工具包
        async with MCPTools(
            command=f"{npx_command} -y @stripe/mcp --tools={enabled_tools} --api-key={stripe_api_key}"
        ) as mcp_toolkit:
            agent = Agent(
                name="StripeAgent",
                instructions=dedent("""\
                    您是专精于 Stripe 操作的 AI 助手。
                    您使用提供的工具与 Stripe API 进行交互。

                    - 理解用户创建或列出 Stripe 对象（客户、产品、价格、支付链接）的请求。
                    - 清晰陈述您操作的结果，包括创建对象的 ID 或检索到的列表。
                    - 如果请求不明确，请请求澄清。
                    - 使用 Markdown 格式，尤其是在链接或代码片段中。
                    - 如果请求涉及多个操作（例如，先创建产品，然后创建价格，最后创建链接），请按顺序执行必要步骤。
                """),
                tools=[mcp_toolkit],
                markdown=True,
                show_tool_calls=True,
            )

            # 使用指定的任务运行代理
            log_info(f"正在使用指派的任务运行代理：'{message}'")
            await agent.aprint_response(message, stream=True)

    except FileNotFoundError:
        error_msg = f"错误：找不到 '{npx_command}' 命令。请确保已在系统 PATH 中安装 Node.js 和 npm/npx。"
        log_error(error_msg)
    except Exception as e:
        log_exception(f"代理执行期间发生意外错误：{e}")


if __name__ == "__main__":
    task = "创建一个名为 'iPhone' 的新 Stripe 产品。然后为其创建一个价格 $999.99 美元。最后，为此价格创建一个支付链接。"
    asyncio.run(run_agent(task))


# 示例提示：
"""
客户管理：
- "创建一个客户。名称：ACME 公司, 邮箱：[e-mail]"
- "列出我的客户。"
- "通过邮箱查找客户 '[e-mail]'" # 注意：这需要 'customers.retrieve' 或搜索功能

产品和价格管理：
- "创建一个名为 'Basic Plan' 的新产品。"
- "为产品 'Basic Plan' 创建一个每月recurring的价格，价格为 $10 美元。"
- "创建一个产品 'Ebook Download' 和一个一次性价格 $19.95 美元。"
- "列出所有产品。" # 注意：这需要 'products.list' 功能
- "列出所有价格。" # 注意：这需要 'prices.list' 功能

支付链接：
- "为每月 $10 美元的 'Basic Plan' 价格创建一个支付链接。"
- "为 '$19.95 Ebook Download' 生成一个支付链接。"

组合任务：
- "创建一个产品 'Pro Service'，添加一个一次性价格 $150 美元，并给我支付链接。"
- "注册一个新客户 '[e-mail]'，名称为 'Support Team'。"
"""