Streamable HTTP Transport ↗
noOriginal Documentation
Documentation Index#
Fetch the complete documentation index at: https://docs.crewai.com/llms.txt Use this file to discover all available pages before exploring further.
Learn how to connect CrewAI to remote MCP servers using the flexible Streamable HTTP transport.
Overview#
Streamable HTTP transport provides a flexible way to connect to remote MCP servers. It’s often built upon HTTP and can support various communication patterns, including request-response and streaming, sometimes utilizing Server-Sent Events (SSE) for server-to-client streams within a broader HTTP interaction.
Key Concepts#
- Remote Servers: Designed for MCP servers hosted remotely.
- Flexibility: Can support more complex interaction patterns than plain SSE, potentially including bi-directional communication if the server implements it.
MCPServerAdapterConfiguration: You’ll need to provide the server’s base URL for MCP communication and specify"streamable-http"as the transport type.
Connecting via Streamable HTTP#
You have two primary methods for managing the connection lifecycle with a Streamable HTTP MCP server:
1. Fully Managed Connection (Recommended)#
The recommended approach is to use a Python context manager (with statement), which handles the connection’s setup and teardown automatically.
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.")Note: Replace "http://localhost:8001/mcp" with the actual URL of your Streamable HTTP MCP server.
2. Manual Connection Lifecycle#
For scenarios requiring more explicit control, you can manage the MCPServerAdapter connection manually.
It is critical to call mcp_server_adapter.stop() when you are done to close the connection and free up resources. A try...finally block is the safest way to ensure this.
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.")Security Considerations#
When using Streamable HTTP transport, general web security best practices are paramount:
- Use HTTPS: Always prefer HTTPS (HTTP Secure) for your MCP server URLs to encrypt data in transit.
- Authentication: Implement robust authentication mechanisms if your MCP server exposes sensitive tools or data.
- Input Validation: Ensure your MCP server validates all incoming requests and parameters.
For a comprehensive guide on securing your MCP integrations, please refer to our Security Considerations page and the official MCP Transport Security documentation.