Spaces:
Paused
Paused
File size: 4,539 Bytes
4b03eed | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 | # -*- coding: utf-8 -*-
"""The MCP client test module in agentscope."""
import asyncio
from multiprocessing import Process
from unittest.async_case import IsolatedAsyncioTestCase
from mcp.server import FastMCP
from mcp.types import EmbeddedResource, TextResourceContents
from agentscope.mcp import MCPClient, HttpMCPConfig
from agentscope.tool import ToolChunk
async def tool_1(arg1: str, arg2: list[int]) -> str:
"""A test tool function.
Args:
arg1 (`str`):
The first argument named arg1.
arg2 (`list[int]`):
The second argument named arg2.
"""
return f"arg1: {arg1}, arg2: {arg2}"
async def tool_2() -> list:
"""
A test tool function return the EmbeddedResource type
"""
return [
EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri="file://tmp.txt",
mimeType="text/plain",
text="test content",
),
),
]
def setup_server() -> None:
"""Set up the streamable HTTP MCP server."""
sse_server = FastMCP("StreamableHTTP", port=8002)
sse_server.tool(description="A test tool function.")(tool_1)
sse_server.tool(
description="A test tool function with embedded resource.",
)(tool_2)
sse_server.run(transport="streamable-http")
class StreamableHttpMCPClientTest(IsolatedAsyncioTestCase):
"""Test class for streamable HTTP MCP client."""
async def asyncTearDown(self) -> None:
"""Tear down the test environment."""
while self.process.is_alive():
self.process.terminate()
await asyncio.sleep(5)
async def asyncSetUp(self) -> None:
"""Set up the test environment."""
self.port = 8002
self.process = Process(target=setup_server)
self.process.start()
await asyncio.sleep(10)
async def test_streamable_http_stateless_client(self) -> None:
"""Test the MCP server connection functionality."""
# Test stateless client (is_stateful=False)
client = MCPClient(
name="test_streamable_http_stateless_client",
is_stateful=False,
mcp_config=HttpMCPConfig(
type="http_mcp",
url=f"http://127.0.0.1:{self.port}/mcp",
),
)
my_tool_1 = await client.get_tool("tool_1")
res_1: ToolChunk = await my_tool_1(arg1="123", arg2=[1, 2, 3])
self.assertEqual(
res_1.content[0].text,
"arg1: 123, arg2: [1, 2, 3]",
)
res_2: ToolChunk = await my_tool_1(arg1="345", arg2=[4, 5, 6])
self.assertEqual(
res_2.content[0].text,
"arg1: 345, arg2: [4, 5, 6]",
)
# Test stateful client (is_stateful=True)
client = MCPClient(
name="test_streamable_http_stateful_client",
is_stateful=True,
mcp_config=HttpMCPConfig(
type="http_mcp",
url=f"http://127.0.0.1:{self.port}/mcp",
),
)
self.assertFalse(client.is_connected)
await client.connect()
self.assertTrue(client.is_connected)
my_tool_1 = await client.get_tool("tool_1")
res_3: ToolChunk = await my_tool_1(arg1="12", arg2=[1, 2])
self.assertEqual(
res_3.content[0].text,
"arg1: 12, arg2: [1, 2]",
)
res_4: ToolChunk = await my_tool_1(arg1="34", arg2=[4, 5])
self.assertEqual(
res_4.content[0].text,
"arg1: 34, arg2: [4, 5]",
)
await client.close()
self.assertFalse(client.is_connected)
async def test_embedded_content(self) -> None:
"""Test the EmbeddedContent functionality."""
# Test with stateless client (is_stateful=False)
client = MCPClient(
name="test_embedded_content",
is_stateful=False,
mcp_config=HttpMCPConfig(
type="http_mcp",
url=f"http://127.0.0.1:{self.port}/mcp",
),
)
my_tool_2 = await client.get_tool("tool_2")
res: ToolChunk = await my_tool_2()
self.assertEqual(
res.content[0].text,
"""{
"uri": "file://tmp.txt/",
"mimeType": "text/plain",
"meta": null,
"text": "test content"
}""",
)
|