跳转到主要内容

​
概述

可流式 HTTP 传输提供了一种连接远程 MCP 服务器的灵活方式。它通常构建在 HTTP 之上，可以支持各种通信模式，包括请求-响应和流式传输，有时利用服务器发送事件 (SSE) 在更广泛的 HTTP 交互中实现服务器到客户端的流。

​
关键概念

  • 远程服务器：专为远程托管的 MCP 服务器设计。
  • 灵活性：可以支持比普通 SSE 更复杂的交互模式，如果服务器实现了双向通信，则可能包括双向通信。
  • MCPServerAdapter 配置：您需要提供服务器用于 MCP 通信的基本 URL，并指定 "streamable-http" 作为传输类型。

​
通过可流式 HTTP 连接

您有两种主要方法来管理与可流式 HTTP MCP 服务器的连接生命周期 推荐的方法是使用 Python 上下文管理器（with 语句），它会自动处理连接的建立和拆除。 
from crewai import Agent, Task, Crew, Process
from crewai_tools import MCPServerAdapter

server_params = {
    "url": "https://: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.")
注意： 请将 "https://:8001/mcp" 替换为您的可流式 HTTP MCP 服务器的实际 URL。

​
2. 手动连接生命周期

对于需要更明确控制的场景，您可以手动管理 MCPServerAdapter 连接。
在完成操作后调用 mcp_server_adapter.stop() 以关闭连接并释放资源是至关重要的。使用 try...finally 代码块是确保此操作得以执行的最安全方式。
from crewai import Agent, Task, Crew, Process
from crewai_tools import MCPServerAdapter

server_params = {
    "url": "https://: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.")

​
安全注意事项

当使用可流式 HTTP 传输时，遵循通用的网络安全最佳实践至关重要
  • 使用 HTTPS：始终优先为您的 MCP 服务器 URL 使用 HTTPS（安全超文本传输协议），以加密传输中的数据。
  • 身份验证：如果您的 MCP 服务器暴露了敏感工具或数据，请实施强大的身份验证机制。
  • 输入验证：确保您的 MCP 服务器验证所有传入的请求和参数。
有关保护您的 MCP 集成的全面指南，请参阅我们的安全注意事项页面和官方 MCP 传输安全文档。
⌘我