9.1 挑战：当任务需要“亿点点”时间

到目前为止，我们所有的MCP工具调用都遵循一个同步的“请求-响应”模式。Host发送一个请求，然后就地等待，直到Server完成任务并返回结果。对于像查询时间、读取小文件或执行简单数据库查询这样的快速任务，这种模式工作得很好。

但是，如果一个任务非常耗时呢？

• 分析一个大型代码库：一个project/analyzeCodebase工具可能需要几分钟来扫描数千个文件。
• 训练一个小型机器学习模型：一个ml/train工具可能需要几十分钟甚至几小时。
• 运行一个复杂的模拟：一个simulation/run工具可能需要数小时才能完成。

在这些情况下，让Host（以及背后的用户）一直阻塞等待是不可接受的。这不仅用户体验极差，而且对于许多系统来说，长时间保持一个HTTP连接或进程通信管道是不可靠的。我们需要一种异步任务处理机制。

异步任务的核心思想：

1. 立即响应 (Immediate Acknowledgement)：当Host请求执行一个长时任务时，Server不再直接执行它。相反，它立即返回一个“任务已收到”的响应，并附带一个唯一的任务ID (Task ID)。
2. 后台执行 (Background Processing)：Server将这个任务放入一个后台任务队列中，由一个或多个工作进程 (Worker) 来异步地执行它。
3. 状态更新与结果通知 (Status Update & Result Notification)：当任务在后台执行时，Host可以（可选地）使用任务ID来查询任务的当前状态（如“排队中”、“运行中”、“已完成”、“失败”）。更重要的是，当任务最终完成时，Server需要一种方式来主动通知Host并交付结果。这就是回调 (Callback) 或 Webhook 机制发挥作用的地方。

在本章中，我们将构建一个long_task_server，它演示了如何使用回调URL来实现这种异步通知机制。这将是MCP协议灵活性和强大扩展性的终极展示。

9.2 异步任务架构设计

为了实现回调，我们的架构需要一些新的组件：

1. Host端需要一个HTTP服务器：Host不再仅仅是一个客户端。它需要暴露一个HTTP端点（回调URL），以便接收来自Server的异步通知。
2. Server端需要一个任务队列：用于管理和执行后台任务。

9.2.1 架构图

9.2.2 项目准备

我们将使用aiohttp库在Host端快速搭建一个简单的HTTP服务器来接收回调。

# 确保你位于ai-research-assistant目录下并激活了虚拟环境
cd ai-research-assistant

# 安装aiohttp
pip install aiohttp

# 创建Long Task Server目录
mkdir long_task_server

# 创建Long Task Server的主程序文件
touch long_task_server/main.py

9.3 构建Long Task MCP Server (long_task_server/main.py)

这个Server的核心是start_long_task工具，它接收一个callback_url参数。

# long_task_server/main.py

import asyncio
import json
import logging
import sys
import uuid
from typing import Dict, Any, List

import httpx

class LongTaskServer:
    def __init__(self):
        self.http_client = httpx.AsyncClient(timeout=30.0)
        logging.basicConfig(level=logging.INFO, format='%(asctime)s - LONG_TASK_SERVER - %(levelname)s - %(message)s')

    async def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
        method = request.get("method")
        params = request.get("params", {})
        request_id = request.get("id")

        try:
            if method == "project/listTools":
                result = self._list_tools()
            elif method == "project/executeTool":
                tool_name = params.get("name")
                tool_params = params.get("parameters", {})
                if tool_name == "task/startLongTask":
                    result = self._start_long_task(tool_params)
                else:
                    raise ValueError(f"Tool '{tool_name}' not found.")
            else:
                raise NotImplementedError(f"Method '{method}' not supported.")
            
            return {"jsonrpc": "2.0", "id": request_id, "result": result}
        except Exception as e:
            logging.error(f"Error handling request: {e}", exc_info=True)
            return {"jsonrpc": "2.0", "id": request_id, "error": {"code": -32000, "message": str(e)}}

    def _list_tools(self) -> List[Dict[str, Any]]:
        return [
            {
                "name": "task/startLongTask",
                "description": "Starts a long-running task and notifies a callback URL upon completion.",
                "parameters": [
                    {"name": "task_description", "type": "string", "required": True, "description": "A description of the task to perform."},
                    {"name": "callback_url", "type": "string", "required": True, "description": "The URL to POST the result to when the task is done."},
                    {"name": "duration_seconds", "type": "integer", "required": False, "description": "Simulated duration of the task.", "default": 10}
                ]
            }
        ]

    def _start_long_task(self, params: Dict[str, Any]) -> Dict[str, Any]:
        task_description = params.get("task_description")
        callback_url = params.get("callback_url")
        duration = params.get("duration_seconds", 10)

        if not callback_url:
            raise ValueError("callback_url is required.")

        task_id = str(uuid.uuid4())
        logging.info(f"Starting long task {task_id}: '{task_description}'. Will notify {callback_url} in {duration}s.")

        # 在后台启动任务，不阻塞当前响应
        asyncio.create_task(self._run_task_in_background(task_id, task_description, callback_url, duration))

        # 立即返回任务ID
        return {
            "result": {"task_id": task_id, "status": "started"},
            "stdout": f"Task {task_id} started. A notification will be sent to {callback_url} upon completion.",
            "stderr": None
        }

    async def _run_task_in_background(self, task_id: str, description: str, callback_url: str, duration: int):
        try:
            # 模拟耗时工作
            await asyncio.sleep(duration)
            
            result_data = f"Task '{description}' completed successfully after {duration} seconds."
            status = "completed"
            logging.info(f"Task {task_id} finished. Notifying {callback_url}.")
        except Exception as e:
            result_data = f"Task '{description}' failed: {e}"
            status = "failed"
            logging.error(f"Task {task_id} failed: {e}", exc_info=True)

        # 任务完成或失败后，调用回调URL
        callback_payload = {
            "task_id": task_id,
            "status": status,
            "result": result_data
        }
        try:
            await self.http_client.post(callback_url, json=callback_payload)
        except httpx.RequestError as e:
            logging.error(f"Failed to send callback for task {task_id} to {callback_url}: {e}")

async def main():
    server = LongTaskServer()
    # 通信循环与之前的Server完全相同
    # ... (此处省略，与之前的Server一致)

if __name__ == "__main__":
    # ... (此处省略，与之前的Server一致)

9.3.1 代码解析

1. _start_long_task 方法: 这是关键。当它被调用时，它不会执行实际的任务。它只做三件事：

   • 生成一个唯一的task_id。
   • 使用asyncio.create_task()将真正的任务执行逻辑（_run_task_in_background）作为一个独立的后台任务来运行。这使得_start_long_task可以立即返回，而不会被阻塞。
   • 立即向Host返回一个包含task_id的响应。

2. _run_task_in_background 方法: 这是一个async方法，它在后台运行。它首先使用asyncio.sleep()来模拟一个耗时的操作。在真实世界中，这里会是CPU密集型或I/O密集型的代码。任务完成后，它会构造一个包含结果的callback_payload，并使用httpx客户端向Host提供的callback_url发送一个POST请求。

3. 无状态设计: 这个Server本身是无状态的。它不存储任务的状态。任务的所有上下文（如callback_url）都在启动时被传递给了后台任务。这种设计使得Server更容易扩展。

9.4 升级Host以处理回调

这是本章最有趣的部分。我们的AiResearchAssistant需要“进化”，从一个纯粹的客户端变成一个同时具有服务器功能的应用。

9.4.1 更新Host主程序 (assistant_host/main.py)

我们需要添加aiohttp服务器，并修改AiResearchAssistant来管理异步任务。

# assistant_host/main.py (重大修改)

import asyncio
import json
import logging
import os
import uuid
from typing import Dict, Any, List

from aiohttp import web

# ... (StdioMcpConnection, McpMultiplexer 类定义保持不变) ...

class AiResearchAssistant:
    def __init__(self, mux: McpMultiplexer, openai_api_key: str, host_port: int):
        self.mux = mux
        self.openai_api_key = openai_api_key
        self.host_port = host_port
        self.callback_url = f"http://localhost:{self.host_port}/mcp_callback"
        self.pending_tasks: Dict[str, Dict[str, Any]] = {}
        logging.basicConfig(level=logging.INFO, format='%(asctime)s - ASSISTANT_HOST - %(levelname)s - %(message)s')

    async def handle_callback(self, request: web.Request) -> web.Response:
        """处理来自MCP Server的异步回调请求"""
        data = await request.json()
        task_id = data.get("task_id")
        logging.info(f"Received callback for task {task_id}: Status - {data.get('status')}")
        if task_id in self.pending_tasks:
            self.pending_tasks[task_id]['status'] = data.get('status')
            self.pending_tasks[task_id]['result'] = data.get('result')
            # 在真实应用中，这里可以触发WebSocket通知、邮件等
            print(f"\n[SYSTEM NOTIFICATION] Task {task_id} finished! Result: {data.get('result')}")
        return web.Response(text="Callback received")

    async def answer(self, question: str) -> str:
        # ... (原有路由逻辑可以保持)

        # 新增：长时任务路由逻辑
        if "run long task" in question.lower():
            params = {
                "name": "task/startLongTask",
                "parameters": {
                    "task_description": question,
                    "callback_url": self.callback_url
                }
            }
            response = await self.mux.request("long_task", "project/executeTool", params)
            task_id = response.get('result', {}).get('result', {}).get('task_id')
            if task_id:
                self.pending_tasks[task_id] = {"status": "started", "result": None}
                return f"Started long-running task with ID: {task_id}. You will be notified upon completion."
            else:
                return "Failed to start the long-running task."

        # ... (原有上下文构建和LLM调用逻辑)
        # ...

async def main():
    # ... (获取API Key)
    HOST_PORT = 8080 # 定义Host监听的端口

    mux = McpMultiplexer()
    # ... (添加所有之前的Server连接)
    long_task_conn = StdioMcpConnection("long_task", "python3 main.py", "../long_task_server")
    mux.add_connection(long_task_conn)

    assistant = AiResearchAssistant(mux, api_key, HOST_PORT)

    # 创建并启动aiohttp服务器来监听回调
    app = web.Application()
    app.router.add_post('/mcp_callback', assistant.handle_callback)
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', HOST_PORT)
    await site.start()
    logging.info(f"Host callback server listening on http://localhost:{HOST_PORT}")

    try:
        await mux.start_all()
        await asyncio.sleep(5)
        logging.info("All servers started. AI Research Assistant is ready.")
        # ... (交互循环)
    finally:
        await runner.cleanup()
        await mux.stop_all()

# ... (主程序入口)

9.4.2 运行与测试

1. 启动Host：

   cd assistant_host
   python3 main.py
   

   你会看到日志显示，除了MCP Server连接，Host自己的HTTP回调服务器也已启动。

2. 发起长时任务：

   Ask your research question (or type 'exit'): Please run long task to analyze the project codebase.
   

3. 观察交互：

   • 立即响应：Host会几乎立刻返回：“Started long-running task with ID: [a-uuid-string]. You will be notified upon completion.” 你可以继续输入其他命令，交互不会被阻塞。
   • 后台活动：在long_task_server的日志中，你会看到它收到了任务，并开始sleep。
   • 回调通知：大约10秒后，在Host的控制台中，会突然打印出一条系统通知：“[SYSTEM NOTIFICATION] Task [a-uuid-string] finished! Result: Task ‘…’ completed successfully after 10 seconds.”
   • Server日志：同时，long_task_server的日志会显示它成功地向回调URL发送了POST请求。

9.5 总结：通向自主Agent的关键一步

掌握了异步任务和回调机制，意味着我们的AI系统真正具备了处理现实世界中复杂、耗时任务的能力。这是从一个简单的“问答机”迈向一个能够独立工作的“自主Agent”的关键一步。

• 解耦执行与监控：Host可以“派发”任务后就去处理其他事情，而不是傻等。这使得构建能够并行处理多个任务、同时响应用户输入的复杂Agent成为可能。
• 协议的弹性：MCP本身没有规定必须是同步的。通过在工具参数中加入callback_url这样的元数据，我们轻松地将协议扩展到了异步场景，展示了其设计的灵活性。
• Agent架构的基础：一个真正的Agent需要一个“主循环”来感知环境、规划任务、执行动作、评估结果。异步回调机制完美地融入了这个循环。Agent可以派发一个动作（一个长时任务），然后在等待回调的间隙里，继续感知环境或规划下一步。

在本教程的最后一章，我们将把前面学到的所有知识——文件、数据库、RAG、Web、异步任务——融会贯通，讨论如何基于MCP构建一个真正的、能够自主规划和执行多步复杂任务的AI Agent。