{{ message }}
Repository navigation
Expand file tree
/
Copy pathtest_extension.py
More file actions
379 lines (267 loc) · 14.3 KB
/
Copy pathtest_extension.py
File metadata and controls
379 lines (267 loc) · 14.3 KB
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
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
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
"""Construction-time tests for `mcp.client.extension`; no session is ever opened."""
from dataclasses import FrozenInstanceError
from typing import Any, Literal, cast
import pytest
from inline_snapshot import snapshot
from mcp_types import CallToolResult, InputRequiredResult, Result
from mcp_types.version import MODERN_PROTOCOL_VERSIONS
from pydantic import AliasChoices, AliasPath, BaseModel, Field
from pydantic.fields import FieldInfo
from mcp.client.extension import (
ClaimContext,
ClientExtension,
NotificationBinding,
ResultClaim,
_wire_keys,
advertise,
)
class _TaskResult(Result):
result_type: Literal["task"] = "task"
task_id: str = "t-1"
class _UntaggedResult(Result):
"""No `result_type` field at all."""
class _PlainStringTagResult(Result):
result_type: str = "task"
class _OtherTagResult(Result):
result_type: Literal["other"] = "other"
class _ClaimedCallToolResult(CallToolResult):
"""A core-result subclass; rejected as a claim model regardless of its tag."""
class _ClaimedInputRequiredResult(InputRequiredResult):
"""A core-result subclass; rejected as a claim model regardless of its tag."""
async def _resolve(result: Result, ctx: ClaimContext) -> CallToolResult:
raise NotImplementedError
def _claim(model: type[Result] = _TaskResult, **kwargs: Any) -> ResultClaim[Result]:
return ResultClaim(result_type="task", model=model, resolve=_resolve, **kwargs)
def test_claim_with_literal_discriminated_model_constructs() -> None:
"""SDK-defined: a model tagged with the claimed Literal constructs, defaulting to `tools/call` everywhere."""
claim = ResultClaim(result_type="task", model=_TaskResult, resolve=_resolve)
assert claim.result_type == "task"
assert claim.model is _TaskResult
assert claim.resolve is _resolve
assert claim.method == "tools/call"
assert claim.protocol_versions is None
def test_claim_accepts_modern_protocol_versions() -> None:
"""SDK-defined: a non-None `protocol_versions` subset of the modern revisions is accepted."""
versions = frozenset(MODERN_PROTOCOL_VERSIONS)
claim = _claim(protocol_versions=versions)
assert claim.protocol_versions == versions
def test_claim_rejects_core_result_type_vocabulary() -> None:
"""SDK-defined: a claim cannot re-key the core tags 'complete' and 'input_required'."""
messages: dict[str, str] = {}
for result_type in ("complete", "input_required"):
with pytest.raises(ValueError) as exc_info:
ResultClaim(result_type=result_type, model=_TaskResult, resolve=_resolve)
messages[result_type] = str(exc_info.value)
assert messages == snapshot(
{
"complete": "resultType 'complete' is core protocol vocabulary",
"input_required": "resultType 'input_required' is core protocol vocabulary",
}
)
@pytest.mark.parametrize("model", [_ClaimedCallToolResult, _ClaimedInputRequiredResult])
def test_claim_rejects_model_subclassing_core_result_types(model: type[Result]) -> None:
"""SDK-defined: a claim model subclassing a core result type is rejected; it would bypass claim routing."""
with pytest.raises(ValueError) as exc_info:
_claim(model=model)
assert str(exc_info.value) == snapshot("claim models must not subclass core result types")
def test_claim_rejects_model_without_result_type_field() -> None:
"""SDK-defined: the claim model must declare the discriminating `result_type` field."""
with pytest.raises(ValueError) as exc_info:
_claim(model=_UntaggedResult)
assert str(exc_info.value) == snapshot("_UntaggedResult.result_type must be Literal['task']")
def test_claim_rejects_plain_str_result_type_field() -> None:
"""SDK-defined: the model's `result_type` must be a Literal of the claimed tag, not a plain `str`."""
with pytest.raises(ValueError) as exc_info:
_claim(model=_PlainStringTagResult)
assert str(exc_info.value) == snapshot("_PlainStringTagResult.result_type must be Literal['task']")
def test_claim_rejects_mismatched_result_type_literal() -> None:
"""SDK-defined: the model's Literal tag must equal the claim's `result_type`."""
with pytest.raises(ValueError) as exc_info:
_claim(model=_OtherTagResult)
assert str(exc_info.value) == snapshot("_OtherTagResult.result_type must be Literal['task']")
class _NotAResult(BaseModel):
result_type: Literal["plain"] = "plain"
class _ReservedAliasResult(Result):
result_type: Literal["clash"] = "clash"
request_state: dict[str, Any] = {}
def test_claim_rejects_model_not_subclassing_result() -> None:
"""SDK-defined: a plain BaseModel cannot be a claim model; the session returns `Result` values."""
with pytest.raises(ValueError) as exc_info:
ResultClaim(result_type="plain", model=cast("type[Result]", _NotAResult), resolve=_resolve)
assert str(exc_info.value) == snapshot("_NotAResult must subclass mcp_types.Result")
def test_claim_rejects_model_aliasing_core_surface_fields() -> None:
"""SDK-defined: a field aliasing requestState or inputRequests would fail core pre-validation."""
with pytest.raises(ValueError) as exc_info:
ResultClaim(result_type="clash", model=_ReservedAliasResult, resolve=_resolve)
assert str(exc_info.value) == snapshot(
"_ReservedAliasResult.request_state aliases 'requestState', a typed field of the core "
"result surface; a colliding value would fail core validation before the claim adapter runs"
)
class _ValidationAliasResult(Result):
result_type: Literal["va"] = "va"
vendor_state: dict[str, Any] | None = Field(default=None, validation_alias="requestState")
class _SerializationAliasResult(Result):
result_type: Literal["sa"] = "sa"
vendor_state: dict[str, Any] | None = Field(default=None, serialization_alias="inputRequests")
class _AliasChoicesResult(Result):
result_type: Literal["ac"] = "ac"
vendor_state: dict[str, Any] | None = Field(
default=None, validation_alias=AliasChoices("vendorKey", "requestState")
)
class _AliasPathResult(Result):
result_type: Literal["ap"] = "ap"
vendor_state: dict[str, Any] | None = Field(
default=None, validation_alias=AliasChoices(AliasPath("requestState", "nested"))
)
def test_wire_keys_for_a_bare_field_is_just_its_name() -> None:
"""SDK-defined: a field with no aliases reads and writes only its own name."""
assert _wire_keys("plain", FieldInfo(annotation=str)) == frozenset({"plain"})
def test_claim_rejects_reserved_aliases_in_every_alias_form() -> None:
"""SDK-defined: validation_alias, serialization_alias, and AliasChoices routes to a reserved key are all caught."""
messages: dict[str, str] = {}
for model in (_ValidationAliasResult, _SerializationAliasResult, _AliasChoicesResult, _AliasPathResult):
with pytest.raises(ValueError) as exc_info:
ResultClaim(result_type=model.model_fields["result_type"].default, model=model, resolve=_resolve)
messages[model.__name__] = str(exc_info.value)
assert messages == snapshot(
{
"_ValidationAliasResult": "_ValidationAliasResult.vendor_state aliases "
"'requestState', a typed field of the core result surface; a colliding value would fail "
"core validation before the claim adapter runs",
"_SerializationAliasResult": "_SerializationAliasResult.vendor_state aliases "
"'inputRequests', a typed field of the core result surface; a colliding value would fail "
"core validation before the claim adapter runs",
"_AliasChoicesResult": "_AliasChoicesResult.vendor_state aliases 'requestState', a typed field of the core "
"result surface; a colliding value would fail core validation before the claim adapter runs",
"_AliasPathResult": "_AliasPathResult.vendor_state aliases "
"'requestState', a typed field of the core result surface; a colliding value would fail "
"core validation before the claim adapter runs",
}
)
def test_claim_rejects_method_outside_the_closed_verb_set() -> None:
"""SDK-defined: claims attach to `tools/call` only, even for values that dodge the static Literal gate."""
with pytest.raises(ValueError) as exc_info:
_claim(method=cast("Literal['tools/call']", "prompts/get"))
assert str(exc_info.value) == snapshot("claims attach to ['tools/call'] only; got method 'prompts/get'")
def test_claim_rejects_empty_protocol_versions() -> None:
"""SDK-defined: an empty version set is rejected; `None` is the spelling for every modern version."""
with pytest.raises(ValueError) as exc_info:
_claim(protocol_versions=frozenset())
assert str(exc_info.value) == snapshot("empty protocol_versions could never activate; use None for all")
def test_claim_rejects_non_modern_protocol_versions() -> None:
"""SDK-defined: a non-None version set must be a subset of the modern protocol revisions."""
messages: list[str] = []
for versions in (
frozenset({"2025-11-25"}),
frozenset({"2026-07-28", "2025-11-25"}),
frozenset({"never-a-version"}),
):
with pytest.raises(ValueError) as exc_info:
_claim(protocol_versions=versions)
messages.append(str(exc_info.value))
assert messages == snapshot(
[
"protocol_versions ['2025-11-25'] are not modern protocol revisions; claimed shapes "
"cannot be delivered on a legacy wire (None means every modern version)",
"protocol_versions ['2025-11-25'] are not modern protocol revisions; claimed shapes "
"cannot be delivered on a legacy wire (None means every modern version)",
"protocol_versions ['never-a-version'] are not modern protocol revisions; claimed shapes "
"cannot be delivered on a legacy wire (None means every modern version)",
]
)
def test_result_claim_is_frozen() -> None:
"""SDK-defined: claims are immutable; mutating one after construction raises."""
claim = _claim()
with pytest.raises(FrozenInstanceError):
setattr(claim, "result_type", "other") # direct assignment is also a type error
class _TaskNotificationParams(BaseModel):
task_id: str
async def _on_task(params: _TaskNotificationParams) -> None:
raise NotImplementedError
def test_notification_binding_constructs() -> None:
"""SDK-defined: a binding is a bare declaration with no construction-time validation."""
binding = NotificationBinding(method="notifications/tasks", params_type=_TaskNotificationParams, handler=_on_task)
assert binding.method == "notifications/tasks"
assert binding.params_type is _TaskNotificationParams
assert binding.handler is _on_task
def test_notification_binding_accepts_core_known_method() -> None:
"""SDK-defined: deliberately no spec-table check at construction, so packages survive core adopting a method."""
binding = NotificationBinding(
method="notifications/progress", params_type=_TaskNotificationParams, handler=_on_task
)
assert binding.method == "notifications/progress"
def test_notification_binding_is_frozen() -> None:
"""SDK-defined: bindings are immutable; mutating one after construction raises."""
binding = NotificationBinding(method="notifications/tasks", params_type=_TaskNotificationParams, handler=_on_task)
with pytest.raises(FrozenInstanceError):
setattr(binding, "method", "notifications/other") # direct assignment is also a type error
def test_extension_defaults_advertise_nothing() -> None:
"""SDK-defined: a minimal subclass advertises empty settings, no claims, and no bindings."""
class _MinimalExt(ClientExtension):
identifier = "com.example/minimal"
ext = _MinimalExt()
assert ext.settings() == {}
assert ext.claims() == ()
assert ext.notifications() == ()
@pytest.mark.parametrize(
"identifier",
[
"io.modelcontextprotocol/ui",
"com.example/my_ext",
"com.x-y.z2/n.a-b_c",
"example/x",
"a/b",
"com.example/9start",
],
)
def test_grammar_conformant_identifiers_accepted_at_class_definition(identifier: str) -> None:
"""Spec `_meta` key grammar: conformant `vendor-prefix/name` identifiers are accepted."""
cls = type("_GoodExt", (ClientExtension,), {"identifier": identifier})
assert cls.identifier == identifier
@pytest.mark.parametrize(
"identifier",
[
"noprefix",
"-foo/bar",
".leading/x",
"a..b/x",
"foo-/x",
"9foo/x",
"foo/-bar",
"foo/bar-",
"foo/",
"/bar",
"foo/ba r",
"io.modelcontextprotocol/ui\n",
"",
42,
],
)
def test_malformed_identifier_rejected_at_class_definition(identifier: Any) -> None:
"""SDK-defined: the SEP-2133 `vendor-prefix/name` grammar is enforced the moment the subclass is defined."""
with pytest.raises(TypeError):
type("_BadExt", (ClientExtension,), {"identifier": identifier})
def test_subclass_without_identifier_allowed_at_definition() -> None:
"""SDK-defined: a subclass with no class-level `identifier` is allowed; validation waits for consumption."""
class _AbstractishExt(ClientExtension):
"""Intermediate base; concrete subclasses supply the identifier."""
class _ConcreteExt(_AbstractishExt):
identifier = "com.example/concrete"
assert _ConcreteExt.identifier == "com.example/concrete"
def test_advertise_serves_captured_settings() -> None:
"""SDK-defined: `advertise()` returns an ad-only extension serving the captured settings."""
ext = advertise("com.example/flags", {"enabled": True})
assert isinstance(ext, ClientExtension)
assert ext.identifier == "com.example/flags"
assert ext.settings() == {"enabled": True}
assert ext.claims() == ()
assert ext.notifications() == ()
def test_advertise_defaults_to_empty_settings() -> None:
"""SDK-defined: omitting settings advertises the extension with an empty map."""
ext = advertise("com.example/flags")
assert ext.settings() == {}
@pytest.mark.parametrize("identifier", ["noprefix", "foo/", ""])
def test_advertise_validates_identifier_eagerly(identifier: str) -> None:
"""SDK-defined: `advertise()` validates the identifier eagerly, at the call site."""
with pytest.raises(TypeError):
advertise(identifier)
You can’t perform that action at this time.