Streamable HTTP 传输

概览

Streamable HTTP 传输提供了一种灵活的方式来连接远程 MCP 服务器。它通常基于 HTTP 构建，并可支持多种通信模式，包括请求-响应和流式传输；在更广泛的 HTTP 交互中，有时还会使用 Server-Sent Events（SSE）进行服务器到客户端的数据流传输。

关键概念

远程服务器：适用于托管在远程的 MCP 服务器。
灵活性：相较于纯 SSE，它可以支持更复杂的交互模式；如果服务器实现了双向通信，也可以支持潜在的双向交互。
MCPServerAdapter 配置：你需要提供用于 MCP 通信的服务器基础 URL，并将传输类型指定为 "streamable-http"。

通过 Streamable HTTP 连接

你有两种主要方式管理与 Streamable HTTP MCP 服务器的连接生命周期：

1. 完全托管连接（推荐）

推荐使用 Python 上下文管理器（with 语句），它会自动处理连接的建立与拆除。

from crewai import Agent, Task, Crew, Process
from crewai_tools import MCPServerAdapter

server_params = {
    "url": "http://localhost:8001/mcp", # Replace with your actual Streamable HTTP server URL
    "transport": "streamable-http"
}

try:
    with MCPServerAdapter(server_params) as tools:
        print(f"Available tools from Streamable HTTP MCP server: {[tool.name for tool in tools]}")

        http_agent = Agent(
            role="HTTP Service Integrator",
            goal="Utilize tools from a remote MCP server via Streamable HTTP.",
            backstory="An AI agent adept at interacting with complex web services.",
            tools=tools,
            verbose=True,
        )

        http_task = Task(
            description="Perform a complex data query using a tool from the Streamable HTTP server.",
            expected_output="The result of the complex data query.",
            agent=http_agent,
        )

        http_crew = Crew(
            agents=[http_agent],
            tasks=[http_task],
            verbose=True,
            process=Process.sequential
        )

        result = http_crew.kickoff()
        print("\nCrew Task Result (Streamable HTTP - Managed):\n", result)

except Exception as e:
    print(f"Error connecting to or using Streamable HTTP MCP server (Managed): {e}")
    print("Ensure the Streamable HTTP MCP server is running and accessible at the specified URL.")

注意： 将 "http://localhost:8001/mcp" 替换为你实际的 Streamable HTTP MCP 服务器 URL。

2. 手动管理连接生命周期

对于需要更明确控制的场景，你可以手动管理 MCPServerAdapter 连接。

from crewai import Agent, Task, Crew, Process
from crewai_tools import MCPServerAdapter

server_params = {
    "url": "http://localhost:8001/mcp", # Replace with your actual Streamable HTTP server URL
    "transport": "streamable-http"
}

mcp_server_adapter = None
try:
    mcp_server_adapter = MCPServerAdapter(server_params)
    mcp_server_adapter.start()
    tools = mcp_server_adapter.tools
    print(f"Available tools (manual Streamable HTTP): {[tool.name for tool in tools]}")

    manual_http_agent = Agent(
        role="Advanced Web Service User",
        goal="Interact with an MCP server using manually managed Streamable HTTP connections.",
        backstory="An AI specialist in fine-tuning HTTP-based service integrations.",
        tools=tools,
        verbose=True
    )

    data_processing_task = Task(
        description="Submit data for processing and retrieve results via Streamable HTTP.",
        expected_output="Processed data or confirmation.",
        agent=manual_http_agent
    )

    data_crew = Crew(
        agents=[manual_http_agent],
        tasks=[data_processing_task],
        verbose=True,
        process=Process.sequential
    )

    result = data_crew.kickoff()
    print("\nCrew Task Result (Streamable HTTP - Manual):\n", result)

except Exception as e:
    print(f"An error occurred during manual Streamable HTTP MCP integration: {e}")
    print("Ensure the Streamable HTTP MCP server is running and accessible.")
finally:
    if mcp_server_adapter and mcp_server_adapter.is_connected:
        print("Stopping Streamable HTTP MCP server connection (manual)...")
        mcp_server_adapter.stop()  # **Crucial: Ensure stop is called**
    elif mcp_server_adapter:
        print("Streamable HTTP MCP server adapter was not connected. No stop needed or start failed.")

安全注意事项

在使用 Streamable HTTP 传输时，应遵循一般的 Web 安全最佳实践：

使用 HTTPS：远程 MCP 服务器 URL 应始终优先使用 HTTPS（HTTP Secure），以加密传输中的数据。
身份验证：如果 MCP 服务器暴露敏感工具或数据，请实现稳健的身份验证机制。
输入验证：确保你的 MCP 服务器会验证所有传入请求和参数。

有关保护 MCP 集成的完整指南，请参阅我们的安全注意事项页面以及官方 MCP Transport Security 文档。