modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Flaky test fix (#1050)

Browse source
This commit is contained in:
AishwaryaKalloli 2025-06-30 02:23:32 +05:30 • committed by GitHub
commit 6f43d1f170
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 15 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

21
tests/issues/test_188_concurrency.py
View file

@ -14,29 +14,38 @@ _resource_name = "slow://slow_resource"
@pytest.mark.anyio @pytest.mark.anyio
async def test_messages_are_executed_concurrently(): async def test_messages_are_executed_concurrently():
server = FastMCP("test") server = FastMCP("test")
call_timestamps = []
@server.tool("sleep") @server.tool("sleep")
async def sleep_tool(): async def sleep_tool():
call_timestamps.append(("tool_start_time", anyio.current_time()))
await anyio.sleep(_sleep_time_seconds) await anyio.sleep(_sleep_time_seconds)
call_timestamps.append(("tool_end_time", anyio.current_time()))
return "done" return "done"
@server.resource(_resource_name) @server.resource(_resource_name)
async def slow_resource(): async def slow_resource():
call_timestamps.append(("resource_start_time", anyio.current_time()))
await anyio.sleep(_sleep_time_seconds) await anyio.sleep(_sleep_time_seconds)
call_timestamps.append(("resource_end_time", anyio.current_time()))
return "slow" return "slow"
async with create_session(server._mcp_server) as client_session: async with create_session(server._mcp_server) as client_session:
start_time = anyio.current_time()
async with anyio.create_task_group() as tg: async with anyio.create_task_group() as tg:
for _ in range(10): for _ in range(10):
tg.start_soon(client_session.call_tool, "sleep") tg.start_soon(client_session.call_tool, "sleep")
tg.start_soon(client_session.read_resource, AnyUrl(_resource_name)) tg.start_soon(client_session.read_resource, AnyUrl(_resource_name))
end_time = anyio.current_time() active_calls = 0
max_concurrent_calls = 0
duration = end_time - start_time for call_type, _ in sorted(call_timestamps, key=lambda x: x[1]):
assert duration < 10 * _sleep_time_seconds if "start" in call_type:
print(duration) active_calls += 1
max_concurrent_calls = max(max_concurrent_calls, active_calls)
else:
active_calls -= 1
print(f"Max concurrent calls: {max_concurrent_calls}")
assert max_concurrent_calls > 1, "No concurrent calls were executed"
def main(): def main():