AI导哈

异步定时器AI代理工具

Resonate示例应用展示了一个结合Resonate和MCP的最小示例，用于实现异步定时器功能。

  • by resonatehq-examples
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 resonatehq-examples/example-agent-tool-background-job

使用说明

异步定时器AI代理工具|共振示例应用程序

此示例显示了谐振+MCP的最小示例。

要在MCP服务器中使用Resonate，需要使用“流式http”作为传输来运行MCP服务器。

因此，您需要一个运行stdio->streaming的http代理，以便Claude与之通信。

代理

from fastmcp import FastMCP

# Create a proxy to a remote server
proxy = FastMCP.as_proxy(
    "http://localhost:5001/mcp",  # URL of the remote server
    name="Remote Server Proxy"
)


if __name__ == "__main__":
    proxy.run()  # Runs via STDIO for Claude Desktop

要运行代理，请执行以下操作：

uv run proxy.py

claude_desktop_config json

确保Claude的配置指向代理。

{
  "mcpServers": {
    "timer": {
      "command": "uv", // or /opt/homebrew/bin/uv
      "args": [
        "--directory",
        "/FULL/PATH/TO/YOUR/TOOL/DIRECTORY/example-agent-tool-async-timer",
        "run",
        "proxy.py"
      ]
    }
  }
}

更改此配置后，您应该重新启动Claude桌面。

在MCP服务器上，确保返回promise ID，而不是阻塞结果。 然后，Claude可以在稍后的任何时候使用promise ID检查结果。

定时器mcp服务器

在后台运行的函数已注册 resonate.register(timer).

定时器函数

async def timer(ctx: Context, timer_name: str, seconds: int) -> str:
    print(f"Timer started: {timer_name} for {seconds} seconds", flush=True)
    await ctx.sleep(int(seconds))
    return "complete"

resonate.register(timer)

Claude与之交互的功能以以下方式装饰 @mcp.tool

设置计时器工具

@mcp.tool()
def set_timer(timer_name: str, seconds: int) -> dict:
    # tool description

    resonate.run(timer_name, timer, timer_name, seconds)
    return {"promise_id": timer_name}

获取计时器状态工具

@mcp.tool()
async def get_timer_status(timer_name: str) -> dict:
    # tool description

    promise_id = f"{timer_name}"
    handle = await resonate.get(promise_id)
    if not handle.done():
        return {"status": "running"}
    return {"status": await handle.result()}

要运行MCP服务器，请执行以下操作：

uv run timer.py

相关 MCP