GeezerVison is showing github.com — python-sdk/tests/docs_src/test_troubleshooting.py at main · modelcontextprotocol/python-sdk

Expand file tree

test_troubleshooting.py

Copy path

More file actions

Latest commit

History

322 lines (262 loc) · 15.9 KB

test_troubleshooting.py

Copy path

File metadata and controls

322 lines (262 loc) · 15.9 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

"""`docs/troubleshooting.md`: every error string the page names, reproduced against the real SDK."""

import logging

from typing import Annotated, Any

import httpx2

import pytest

from inline_snapshot import snapshot

from mcp_types import (

INVALID_PARAMS,

INVALID_REQUEST,

MISSING_REQUIRED_CLIENT_CAPABILITY,

ElicitRequestParams,

ElicitResult,

ErrorData,

TextContent,

)

from pydantic import Field

from docs_src.troubleshooting import (

tutorial001,

tutorial002,

tutorial003,

tutorial004,

tutorial005,

tutorial006,

tutorial007,

tutorial008,

)

from mcp import Client, MCPError

from mcp.client import ClientRequestContext

from mcp.client.streamable_http import streamable_http_client

from mcp.server import MCPServer

from mcp.server.mcpserver import RequestStateSecurity

from mcp.server.mcpserver.exceptions import InvalidSignature

# See test_index.py for why this is a per-module mark and not a conftest hook.

pytestmark = [pytest.mark.anyio, pytest.mark.filterwarnings("error::mcp.MCPDeprecationWarning")]

INITIALIZE = {

"jsonrpc": "2.0",

"id": 1,

"method": "initialize",

"params": {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "b", "version": "1"}},

}

MCP_HEADERS = {"Accept": "application/json, text/event-stream", "Content-Type": "application/json"}

async def _confirm(context: ClientRequestContext, params: ElicitRequestParams) -> ElicitResult:

"""The page's one `elicitation_callback`: always accept the booking."""

return ElicitResult(action="accept", content={"confirm": True})

async def test_an_error_leaving_the_async_with_block_arrives_wrapped_in_an_exception_group() -> None:

"""The `unhandled errors in a TaskGroup` entry: anyio group-wraps whatever escapes the block."""

with pytest.raises(Exception) as exc_info:

async with Client(tutorial001.mcp) as client:

await client.read_resource("weather://Atlantis")

assert not isinstance(exc_info.value, MCPError)

assert exc_info.group_contains(MCPError, match=r"^No forecast for 'Atlantis'\.$")

async def test_the_same_error_caught_inside_the_block_is_the_bare_mcp_error() -> None:

"""The fix on the page: `except MCPError` inside the `async with` never sees an `ExceptionGroup`."""

async with Client(tutorial001.mcp) as client:

with pytest.raises(MCPError) as exc_info:

await client.read_resource("weather://Atlantis")

assert str(exc_info.value) == "No forecast for 'Atlantis'."

assert exc_info.value.error.code == INVALID_PARAMS

async def test_a_client_outside_its_async_with_refuses_every_call() -> None:

"""`Client(...)` only constructs. Nothing connects until `async with`, so every call refuses."""

client = Client(tutorial001.mcp)

with pytest.raises(RuntimeError, match="^Client must be used within an async context manager$"):

await client.list_tools()

async def test_a_failing_tool_returns_is_error_true_instead_of_raising() -> None:

"""The `Error executing tool` entry: it is a result, not an exception. Nothing to `except`."""

async with Client(tutorial001.mcp) as client:

result = await client.call_tool("forecast", {"city": "Atlantis"})

assert result.is_error

assert result.content == [

TextContent(type="text", text="Error executing tool forecast: No forecast for 'Atlantis'.")

]

async def test_a_crashing_tool_is_the_bare_form_with_its_traceback_in_the_server_log(

caplog: pytest.LogCaptureFixture,

) -> None:

"""The bare `Error executing tool <name>`: the tool raised something other than ToolError, its text

is withheld, and the server log carries the exact ERROR message the page names."""

mcp = MCPServer("Weather")

@mcp.tool()

def forecast(city: str) -> str:

"""Today's forecast for one city. Crashes: the upstream table is missing the key."""

forecasts: dict[str, str] = {}

return forecasts[city]

caplog.set_level(logging.ERROR, logger="mcp.server.mcpserver.server")

async with Client(mcp) as client:

result = await client.call_tool("forecast", {"city": "Atlantis"})

assert result.content == [TextContent(type="text", text="Error executing tool forecast")]

(record,) = [r for r in caplog.records if r.name == "mcp.server.mcpserver.server"]

assert record.getMessage() == "Tool 'forecast' raised an unexpected exception"

assert record.exc_info is not None

async def test_an_unknown_tool_is_the_same_kind_of_result() -> None:

"""`Unknown tool: <name>` travels the same `is_error=True` path as a failing tool."""

async with Client(tutorial001.mcp) as client:

result = await client.call_tool("get_forecast", {"city": "London"})

assert result.is_error

assert result.content == [TextContent(type="text", text="Unknown tool: get_forecast")]

async def test_the_tool_decorator_without_parentheses_raises_at_import_time() -> None:

"""`@mcp.tool` (no parentheses) hands the function itself to `name=`; the SDK refuses immediately."""

mcp = MCPServer("Weather")

undecorated: Any = mcp.tool

with pytest.raises(TypeError, match=r"Use @tool instead of @tool"):

@undecorated

def forecast(city: str) -> None:

"""Today's forecast for one city. Never called: the decoration itself is what raises."""

async def test_an_invalid_x_mcp_header_annotation_raises_at_import_time() -> None:

"""A marked `str | None` is refused by the decorator itself, with the heading's text and the reason after it."""

mcp = MCPServer("Weather")

with pytest.raises(InvalidSignature) as excinfo:

@mcp.tool()

def forecast(city: Annotated[str | None, Field(json_schema_extra={"x-mcp-header": "City"})] = None) -> None:

"""Today's forecast for one city. Never called: the decoration itself is what raises."""

assert str(excinfo.value) == snapshot(

"Tool 'forecast' has an invalid x-mcp-header annotation: "

"property 'city': x-mcp-header is only permitted on integer/string/boolean properties "

"(the type keyword is NoneType, not a string)"

)

async def test_a_duplicate_tool_name_keeps_the_first_and_drops_the_second() -> None:

"""tutorial002: `tools/list` reports one `forecast`, and it is the first registration that won."""

async with Client(tutorial002.mcp) as client:

(tool,) = (await client.list_tools()).tools

assert tool.name == "forecast"

assert tool.description == "Today's forecast for one city."

async def test_a_duplicate_registration_logs_tool_already_exists(caplog: pytest.LogCaptureFixture) -> None:

"""The only signal for a dropped duplicate is the `Tool already exists:` warning in the server log."""

with caplog.at_level(logging.WARNING, logger="mcp.server.mcpserver.tools.tool_manager"):

@tutorial002.mcp.tool(name="forecast")

def forecast_weekly(city: str) -> None:

"""The week ahead for one city. Never called: it is the duplicate that gets dropped."""

assert "Tool already exists: forecast" in caplog.messages

async def test_the_default_streamable_http_app_answers_a_real_hostname_with_421(

caplog: pytest.LogCaptureFixture,

) -> None:

"""tutorial003: one 421, three spellings. The page presents all three as the same event."""

transport = httpx2.ASGITransport(app=tutorial003.app)

async with tutorial003.mcp.session_manager.run():

# What curl (or the reverse proxy's access log) shows: the status and the plain-text body.

async with httpx2.AsyncClient(transport=transport, base_url="http://mcp.example.com") as raw:

with caplog.at_level(logging.WARNING, logger="mcp.server.transport_security"):

response = await raw.post("/mcp", json=INITIALIZE, headers=MCP_HEADERS)

assert (response.status_code, response.text) == (421, "Invalid Host header")

# No `Content-Type: application/json`, which is exactly why the python client cannot show the body.

assert response.headers.get("content-type") is None

# What the server operator finds by grepping the server log.

assert "Invalid Host header: mcp.example.com" in caplog.messages

# What the python `Client` raises instead: the generic stand-in, wrapped by the task group.

async with httpx2.AsyncClient(transport=transport) as http_client:

client = Client(streamable_http_client("http://mcp.example.com/mcp", http_client=http_client))

with pytest.raises(Exception) as exc_info: # pragma: no branch

await client.__aenter__() # the connection attempt itself is what fails

assert not isinstance(exc_info.value, MCPError)

assert exc_info.group_contains(MCPError, match="^Server returned an error response$")

async def test_an_allowlisted_hostname_connects_and_calls_a_tool() -> None:

"""tutorial004: `transport_security=` names the deployed hostname, and the same client connects."""

transport = httpx2.ASGITransport(app=tutorial004.app)

async with tutorial004.mcp.session_manager.run():

async with httpx2.AsyncClient(transport=transport) as http_client:

allowed = streamable_http_client("http://mcp.example.com/mcp", http_client=http_client)

async with Client(allowed) as c: # pragma: no branch

assert c.protocol_version == "2026-07-28"

result = await c.call_tool("forecast", {"city": "London"})

assert result.structured_content == {"result": "London: Rain."}

async def test_a_mounted_app_without_a_lifespan_fails_on_the_first_request() -> None:

"""tutorial005: Starlette never runs a mounted sub-app's lifespan, so nothing starts the manager."""

transport = httpx2.ASGITransport(app=tutorial005.app)

async with httpx2.AsyncClient(transport=transport, base_url="http://127.0.0.1:8000") as http:

with pytest.raises(RuntimeError, match=r"Task group is not initialized\. Make sure to use run\."):

await http.post("/mcp")

async def test_a_session_id_the_server_never_issued_gets_a_404_session_not_found() -> None:

"""`Session not found` is a 404 with a JSON-RPC body, so the python `Client` surfaces it verbatim."""

mcp = MCPServer("Weather")

app = mcp.streamable_http_app()

async with mcp.session_manager.run():

async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app), base_url="http://127.0.0.1:8000") as h:

response = await h.post(

"/mcp",

json={"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}},

headers={**MCP_HEADERS, "mcp-session-id": "deadbeef"},

)

assert response.status_code == 404

assert response.headers["content-type"] == "application/json"

assert response.json() == {"jsonrpc": "2.0", "id": None, "error": {"code": -32600, "message": "Session not found"}}

async def test_ctx_elicit_at_2026_has_no_back_channel() -> None:

"""tutorial006: at 2026-07-28 the server refuses to send `elicitation/create` at all."""

async with Client(tutorial006.mcp) as client:

assert client.protocol_version == "2026-07-28"

with pytest.raises(MCPError) as exc_info:

await client.call_tool("book_table", {"date": "Friday"})

assert exc_info.value.error == ErrorData(

code=INVALID_REQUEST,

message=(

"Cannot send 'elicitation/create': "

"this transport context has no back-channel for server-initiated requests."

)

async def test_an_elicitation_callback_does_not_fix_ctx_elicit_at_2026() -> None:

"""The page's claim: registering the callback changes nothing. No request ever reaches the client."""

async with Client(tutorial006.mcp, elicitation_callback=_confirm) as client:

with pytest.raises(MCPError, match="no back-channel for server-initiated requests"):

await client.call_tool("book_table", {"date": "Friday"})

async def test_ctx_elicit_on_a_legacy_connection_works() -> None:

"""The legacy aside: `ctx.elicit` is a server-to-client request, and only a legacy session has those."""

async with Client(tutorial006.mcp, mode="legacy", elicitation_callback=_confirm) as client:

result = await client.call_tool("book_table", {"date": "Friday"})

assert result.structured_content == {"result": "Booked for Friday."}

async def test_the_resolver_form_works_on_a_2026_connection() -> None:

"""tutorial007: the fix. Same question, same callback, but the server returns it instead of calling back."""

async with Client(tutorial007.mcp, elicitation_callback=_confirm) as client:

assert client.protocol_version == "2026-07-28"

result = await client.call_tool("book_table", {"date": "Friday"})

assert result.structured_content == {"result": "Booked for Friday."}

async def test_the_resolver_form_without_a_callback_names_the_missing_capability() -> None:

"""The `-32021` entry: the server refuses up front, and `data` names the capability to declare."""

async with Client(tutorial007.mcp) as client:

with pytest.raises(MCPError) as exc_info:

await client.call_tool("book_table", {"date": "Friday"})

assert exc_info.value.error == ErrorData(

code=MISSING_REQUIRED_CLIENT_CAPABILITY,

message=(

"Client did not declare the form elicitation capability required by resolver "

"'docs_src.troubleshooting.tutorial007:ask_to_confirm'"

data={"requiredCapabilities": {"elicitation": {"form": {}}}},

)

async def test_a_legacy_ctx_elicit_without_a_callback_says_elicitation_not_supported() -> None:

"""The `Elicitation not supported` entry: no `elicitation_callback` means nobody to ask."""

async with Client(tutorial006.mcp, mode="legacy") as client:

with pytest.raises(MCPError) as exc_info:

await client.call_tool("book_table", {"date": "Friday"})

assert exc_info.value.error == ErrorData(code=INVALID_REQUEST, message="Elicitation not supported")

async def test_ctx_elicit_over_stateless_http_has_no_back_channel() -> None:

"""tutorial008: `stateless_http=True` leaves the server no channel to send `elicitation/create`."""

transport = httpx2.ASGITransport(app=tutorial008.app)

async with tutorial008.mcp.session_manager.run():

async with httpx2.AsyncClient(transport=transport) as http_client:

stateless = streamable_http_client("http://127.0.0.1:8000/mcp", http_client=http_client)

async with Client(stateless) as c: # pragma: no branch

with pytest.raises(MCPError) as exc_info: # pragma: no branch

await c.call_tool("book_table", {"date": "Friday"})

assert exc_info.value.error == ErrorData(

code=INVALID_REQUEST,

message=(

"Cannot send 'elicitation/create': "

"this transport context has no back-channel for server-initiated requests."

)

async def test_a_request_state_the_server_did_not_mint_is_rejected(caplog: pytest.LogCaptureFixture) -> None:

"""The wire message is deliberately frozen; the real reason goes only to the server log."""

async with Client(tutorial001.mcp) as client:

with caplog.at_level(logging.WARNING, logger="mcp.server.request_state"):

with pytest.raises(MCPError) as exc_info: # pragma: no branch

await client.call_tool("forecast", {"city": "London"}, request_state="round-1-from-worker-a")

assert exc_info.value.error == ErrorData(

code=INVALID_PARAMS, message="Invalid or expired requestState", data={"reason": "invalid_request_state"}

)

assert "requestState rejected on tools/call: malformed" in caplog.messages

async def test_a_short_request_state_key_is_rejected_at_construction() -> None:

"""`RequestStateSecurity(keys=[...])` refuses anything under 32 bytes and says how to make one."""

with pytest.raises(ValueError) as exc_info:

RequestStateSecurity(keys=[b"hunter2"])

assert str(exc_info.value) == (

"request-state keys must be at least 32 bytes of secret randomness; keys[0] is 7 bytes. "

'Generate one with: python -c "import secrets; print(secrets.token_hex(32))"'

)