MCPStreamableHTTPTool Class

MCP tool for connecting to HTTP-based MCP servers.

This class connects to MCP servers that communicate via streamable HTTP/SSE.

Initialize the MCP streamable HTTP tool.

Note

The arguments are used to create a streamable HTTP client.

See mcp.client.streamable_http.streamablehttp_client for more details.

Any extra arguments passed to the constructor will be passed to the

streamable HTTP client constructor.

Constructor

MCPStreamableHTTPTool()

Parameters

Name	Description
name Required	The name of the tool.
url Required	The URL of the MCP server.

Keyword-Only Parameters

Name	Description
load_tools	Whether to load tools from the MCP server. Default value: True
load_prompts	Whether to load prompts from the MCP server. Default value: True
request_timeout	The default timeout in seconds for all requests. Default value: None
session	The session to use for the MCP connection. Default value: None
description	The description of the tool. Default value: None
approval_mode	The approval mode for the tool. This can be: "always_require": The tool always requires approval before use. "never_require": The tool never requires approval before use. A dict with keys always_require_approval or never_require_approval, followed by a sequence of strings with the names of the relevant tools. A tool should not be listed in both, if so, it will require approval. Default value: None
allowed_tools	A list of tools that are allowed to use this tool. Default value: None
additional_properties	Additional properties. Default value: None
headers	The headers to send with the request. Default value: None
timeout	The timeout for the request. Default value: None
sse_read_timeout	The timeout for reading from the SSE stream. Default value: None
terminate_on_close	Close the transport when the MCP client is terminated. Default value: None
chat_client	The chat client to use for sampling. Default value: None
kwargs	Any extra arguments to pass to the SSE client.

Examples


   from agent_framework import MCPStreamableHTTPTool, ChatAgent

   # Create an MCP HTTP tool
   mcp_tool = MCPStreamableHTTPTool(
       name="web-api",
       url="https://api.example.com/mcp",
       headers={"Authorization": "Bearer token"},
       description="Web API operations",
   )

   # Use with a chat agent
   async with mcp_tool:
       agent = ChatAgent(chat_client=client, name="assistant", tools=mcp_tool)
       response = await agent.run("Fetch data from the API")

Methods

__init__	Initialize the MCP streamable HTTP tool. Note The arguments are used to create a streamable HTTP client. See mcp.client.streamable_http.streamablehttp_client for more details. Any extra arguments passed to the constructor will be passed to the streamable HTTP client constructor.
__new__
call_tool	Call a tool with the given arguments.
close	Disconnect from the MCP server. Closes the connection and cleans up resources.
connect	Connect to the MCP server. Establishes a connection to the MCP server, initializes the session, and loads tools and prompts if configured to do so.
get_mcp_client	Get an MCP streamable HTTP client.
get_prompt	Call a prompt with the given arguments.
load_prompts	Load prompts from the MCP server. Retrieves available prompts from the connected MCP server and converts them into AIFunction instances.
load_tools	Load tools from the MCP server. Retrieves available tools from the connected MCP server and converts them into AIFunction instances.
logging_callback	Callback function for logging. This function is called when the MCP Server sends a log message. By default it will log the message to the logger with the level set in the params. Note Subclass MCPTool and override this function if you want to adapt the behavior.
message_handler	Handle messages from the MCP server. By default this function will handle exceptions on the server by logging them, and it will trigger a reload of the tools and prompts when the list changed notification is received. Note If you want to extend this behavior, you can subclass MCPTool and override this function. If you want to keep the default behavior, make sure to call super().message_handler(message).
sampling_callback	Callback function for sampling. This function is called when the MCP server needs to get a message completed. It uses the configured chat client to generate responses. Note This is a simple version of this function. It can be overridden to allow more complex sampling. It gets added to the session at initialization time, so overriding it is the best way to customize this behavior.

init

Initialize the MCP streamable HTTP tool.

Note

The arguments are used to create a streamable HTTP client.

See mcp.client.streamable_http.streamablehttp_client for more details.

Any extra arguments passed to the constructor will be passed to the

streamable HTTP client constructor.

__init__(name: str, url: str, *, load_tools: bool = True, load_prompts: bool = True, request_timeout: int | None = None, session: ClientSession | None = None, description: str | None = None, approval_mode: Literal['always_require', 'never_require'] | HostedMCPSpecificApproval | None = None, allowed_tools: Collection[str] | None = None, headers: dict[str, Any] | None = None, timeout: float | None = None, sse_read_timeout: float | None = None, terminate_on_close: bool | None = None, chat_client: ChatClientProtocol | None = None, additional_properties: dict[str, Any] | None = None, **kwargs: Any) -> None

Parameters

Name	Description
name Required	str The name of the tool.
url Required	str The URL of the MCP server.
load_tools Required	bool
load_prompts Required	bool
request_timeout Required	int \| None
session Required	<xref:mcp.client.session.ClientSession> \| None
description Required	str \| None
approval_mode Required	Literal['always_require', 'never_require'] \| agent_framework._tools.HostedMCPSpecificApproval \| None
allowed_tools Required	Collection[str] \| None
headers Required	dict[str, Any] \| None
timeout Required	float \| None
sse_read_timeout Required	float \| None
terminate_on_close Required	bool \| None
chat_client Required	ChatClientProtocol \| None
additional_properties Required	dict[str, Any] \| None
kwargs Required	Any

Keyword-Only Parameters

Name	Description
load_tools	Whether to load tools from the MCP server. Default value: True
load_prompts	Whether to load prompts from the MCP server. Default value: True
request_timeout	The default timeout in seconds for all requests. Default value: None
session	The session to use for the MCP connection. Default value: None
description	The description of the tool. Default value: None
approval_mode	The approval mode for the tool. This can be: "always_require": The tool always requires approval before use. "never_require": The tool never requires approval before use. A dict with keys always_require_approval or never_require_approval, followed by a sequence of strings with the names of the relevant tools. A tool should not be listed in both, if so, it will require approval. Default value: None
allowed_tools	A list of tools that are allowed to use this tool. Default value: None
additional_properties	Additional properties. Default value: None
headers	The headers to send with the request. Default value: None
timeout	The timeout for the request. Default value: None
sse_read_timeout	The timeout for reading from the SSE stream. Default value: None
terminate_on_close	Close the transport when the MCP client is terminated. Default value: None
chat_client	The chat client to use for sampling. Default value: None
kwargs	Any extra arguments to pass to the SSE client.

Returns

Type	Description
None

new

__new__(**kwargs)

call_tool

Call a tool with the given arguments.

async call_tool(tool_name: str, **kwargs: Any) -> list[TextContent | DataContent | TextReasoningContent | UriContent | FunctionCallContent | FunctionResultContent | ErrorContent | UsageContent | HostedFileContent | HostedVectorStoreContent | FunctionApprovalRequestContent | FunctionApprovalResponseContent]

Parameters

Name	Description
tool_name Required	str The name of the tool to call.
kwargs Required	Any

Keyword-Only Parameters

Name	Description
kwargs	Arguments to pass to the tool.

Returns

Type	Description
list[TextContent \| DataContent \| TextReasoningContent \| UriContent \| FunctionCallContent \| FunctionResultContent \| ErrorContent \| UsageContent \| HostedFileContent \| HostedVectorStoreContent \| FunctionApprovalRequestContent \| FunctionApprovalResponseContent]	A list of content items returned by the tool.

Exceptions

Type	Description
ToolExecutionException	If the MCP server is not connected, tools are not loaded, or the tool call fails.

close

Disconnect from the MCP server.

Closes the connection and cleans up resources.

async close() -> None

Returns

Type	Description
None

connect

Connect to the MCP server.

Establishes a connection to the MCP server, initializes the session, and loads tools and prompts if configured to do so.

async connect() -> None

Returns

Type	Description
None

Exceptions

Type	Description
ToolException	If connection or session initialization fails.

get_mcp_client

Get an MCP streamable HTTP client.

get_mcp_client() -> _AsyncGeneratorContextManager[Any, None]

Returns

Type	Description
<xref:contextlib._AsyncGeneratorContextManager>[Any, None]	An async context manager for the streamable HTTP client transport.

get_prompt

Call a prompt with the given arguments.

async get_prompt(prompt_name: str, **kwargs: Any) -> list[ChatMessage]

Parameters

Name	Description
prompt_name Required	str The name of the prompt to retrieve.
kwargs Required	Any

Keyword-Only Parameters

Name	Description
kwargs	Arguments to pass to the prompt.

Returns

Type	Description
list[ChatMessage]	A list of chat messages returned by the prompt.

Exceptions

Type	Description
ToolExecutionException	If the MCP server is not connected, prompts are not loaded, or the prompt call fails.

load_prompts

Load prompts from the MCP server.

Retrieves available prompts from the connected MCP server and converts them into AIFunction instances.

async load_prompts() -> None

Returns

Type	Description
None

Exceptions

Type	Description
ToolExecutionException	If the MCP server is not connected.

load_tools

Load tools from the MCP server.

Retrieves available tools from the connected MCP server and converts them into AIFunction instances.

async load_tools() -> None

Returns

Type	Description
None

Exceptions

Type	Description
ToolExecutionException	If the MCP server is not connected.

logging_callback

Callback function for logging.

This function is called when the MCP Server sends a log message. By default it will log the message to the logger with the level set in the params.

Note

Subclass MCPTool and override this function if you want to adapt the behavior.

async logging_callback(params: LoggingMessageNotificationParams) -> None

Parameters

Name	Description
params Required	<xref:mcp.types.LoggingMessageNotificationParams> The logging message notification parameters from the MCP server.

Returns

Type	Description
None

message_handler

Handle messages from the MCP server.

By default this function will handle exceptions on the server by logging them, and it will trigger a reload of the tools and prompts when the list changed notification is received.

Note

If you want to extend this behavior, you can subclass MCPTool and override

this function. If you want to keep the default behavior, make sure to call

super().message_handler(message).

async message_handler(message: RequestResponder[ServerRequest, ClientResult] | ServerNotification | Exception) -> None

Parameters

Name	Description
message Required	<xref:mcp.shared.session.RequestResponder>[<xref:mcp.types.ServerRequest>, <xref:mcp.types.ClientResult>] \| <xref:mcp.types.ServerNotification> \| Exception The message from the MCP server (request responder, notification, or exception).

Returns

Type	Description
None

sampling_callback

Callback function for sampling.

This function is called when the MCP server needs to get a message completed. It uses the configured chat client to generate responses.

Note

This is a simple version of this function. It can be overridden to allow

more complex sampling. It gets added to the session at initialization time,

so overriding it is the best way to customize this behavior.

async sampling_callback(context: RequestContext[ClientSession, Any, Any], params: CreateMessageRequestParams) -> CreateMessageResult | ErrorData

Parameters

Name	Description
context Required	<xref:mcp.shared.context.RequestContext>[<xref:mcp.client.session.ClientSession>, Any, Any] The request context from the MCP server.
params Required	<xref:mcp.types.CreateMessageRequestParams> The message creation request parameters.

Returns

Type	Description
<xref:mcp.types.CreateMessageResult> \| <xref:mcp.types.ErrorData>	Either a CreateMessageResult with the generated message or ErrorData if generation fails.

MCPStreamableHTTPTool Class

Constructor

Parameters

Keyword-Only Parameters

Examples

Methods

__init__

Parameters

Keyword-Only Parameters

Returns

__new__

call_tool

Parameters

Keyword-Only Parameters

Returns

Exceptions

close

Returns

connect

Returns

Exceptions

get_mcp_client

Returns

get_prompt

Parameters

Keyword-Only Parameters

Returns

Exceptions

load_prompts

Returns

Exceptions

load_tools

Returns

Exceptions

logging_callback

Parameters

Returns

message_handler

Parameters

Returns

sampling_callback

Parameters

Returns

Attributes

functions

Feedback

init

new