19
19
import subprocess
20
20
import sys
21
21
import time
22
- from collections .abc import AsyncGenerator
23
- from contextlib import asynccontextmanager
24
22
25
23
import httpx
26
24
import pytest
27
- import uvicorn
28
- from starlette .applications import Starlette
29
- from starlette .routing import Mount
30
- from starlette .types import Receive , Scope , Send
31
-
32
- from mcp .server import Server
33
- from mcp .server .streamable_http_manager import StreamableHTTPSessionManager
34
- from mcp .types import Tool
35
25
36
26
SERVER_NAME = "test_race_condition_server"
37
27
@@ -85,16 +75,40 @@ def server_url(server_port: int) -> str:
85
75
return f"http://127.0.0.1:{ server_port } "
86
76
87
77
78
+ def start_server_process (port : int ) -> subprocess .Popen [str ]:
79
+ """Start server in a separate process."""
80
+ # Create a temporary script to run the server
81
+ import os
82
+
83
+ server_code = f"""
84
+ import sys
85
+ import os
86
+ sys.path.insert(0, { repr (os .getcwd ())} )
87
+
88
+ import socket
89
+ import time
90
+ from collections.abc import AsyncGenerator
91
+ from contextlib import asynccontextmanager
92
+
93
+ import uvicorn
94
+ from starlette.applications import Starlette
95
+ from starlette.routing import Mount
96
+ from starlette.types import Receive, Scope, Send
97
+
98
+ from mcp.server import Server
99
+ from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
100
+ from mcp.types import Tool
101
+
102
+ SERVER_NAME = "test_race_condition_server"
103
+
88
104
class RaceConditionTestServer(Server):
89
105
def __init__(self):
90
106
super().__init__(SERVER_NAME)
91
107
92
108
async def on_list_tools(self) -> list[Tool]:
93
109
return []
94
110
95
-
96
111
def run_server_with_logging(port: int) -> None:
97
- """Run the StreamableHTTP server with logging to capture race condition errors."""
98
112
app = RaceConditionTestServer()
99
113
100
114
# Create session manager
@@ -121,16 +135,6 @@ async def lifespan(app: Starlette) -> AsyncGenerator[None, None]:
121
135
starlette_app = Starlette(routes=routes, lifespan=lifespan)
122
136
uvicorn.run(starlette_app, host="127.0.0.1", port=port, log_level="debug")
123
137
124
-
125
- def start_server_process (port : int ) -> subprocess .Popen [str ]:
126
- """Start server in a separate process."""
127
- # Create a temporary script to run the server
128
- import os
129
-
130
- server_code = f"""
131
- import sys
132
- sys.path.insert(0, { repr (os .getcwd ())} )
133
- from tests.issues.test_1363_race_condition_streamable_http import run_server_with_logging
134
138
run_server_with_logging({ port } )
135
139
"""
136
140
0 commit comments