encounter/adk-python
0
0
Fork 0
mirror of https://github.com/encounter/adk-python.git synced 2026-07-09 18:19:28 -07:00
Code Issues Packages Projects Releases Wiki Activity

Feat/expose mcps streamable http custom httpx factory parameter (#2997)

Browse Source 
* feat: Add support for custom HTTPX client factory in StreamableHTTPConnectionParams

* Update src/google/adk/tools/mcp_tool/mcp_session_manager.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

* unit tested mock

* provide default - httpx client factory can't be none

* feat: Enhance StreamableHTTPConnectionParams with httpx_client_factory attribute

* fmt

* fmt

* refactor: Rename test_init_with_streamable_http_none_httpx_factory to test_init_with_streamable_http_default_httpx_factory for clarity

* isort

* fmt

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
Co-authored-by: Kathy Wu <108756731+wukath@users.noreply.github.com>
This commit is contained in:
Michael Jones
2025-11-18 11:09:48 -08:00
committed by GitHub
co-authored by gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> Kathy Wu
parent b5f5df9fa8
commit 0cc3d6d6d5
2 changed files with 69 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unittests/tools/mcp_tool/test_mcp_session_manager.py
+53
View File
@@ -146,6 +146,59 @@ class TestMCPSessionManager:
assert manager._connection_params == http_params
@patch("google.adk.tools.mcp_tool.mcp_session_manager.streamablehttp_client")
def test_init_with_streamable_http_custom_httpx_factory(
self, mock_streamablehttp_client
):
"""Test that streamablehttp_client is called with custom httpx_client_factory."""
from datetime import timedelta
custom_httpx_factory = Mock()
http_params = StreamableHTTPConnectionParams(
url="https://example.com/mcp",
timeout=15.0,
httpx_client_factory=custom_httpx_factory,
)
manager = MCPSessionManager(http_params)
manager._create_client()
mock_streamablehttp_client.assert_called_once_with(
url="https://example.com/mcp",
headers=None,
timeout=timedelta(seconds=15.0),
sse_read_timeout=timedelta(seconds=300.0),
terminate_on_close=True,
httpx_client_factory=custom_httpx_factory,
)
@pytest.mark.asyncio
@patch("google.adk.tools.mcp_tool.mcp_session_manager.streamablehttp_client")
async def test_init_with_streamable_http_default_httpx_factory(
self, mock_streamablehttp_client
):
"""Test that streamablehttp_client is called with custom httpx_client_factory."""
from datetime import timedelta
from mcp.client.streamable_http import create_mcp_http_client
http_params = StreamableHTTPConnectionParams(
url="https://example.com/mcp", timeout=15.0
)
manager = MCPSessionManager(http_params)
manager._create_client()
mock_streamablehttp_client.assert_called_once_with(
url="https://example.com/mcp",
headers=None,
timeout=timedelta(seconds=15.0),
sse_read_timeout=timedelta(seconds=300.0),
terminate_on_close=True,
httpx_client_factory=create_mcp_http_client,
)
def test_generate_session_key_stdio(self):
"""Test session key generation for stdio connections."""
manager = MCPSessionManager(self.mock_stdio_connection_params)