FilesExpand file tree

v2.0.1

/

test_server_context.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

126 lines (102 loc) · 5 KB

v2.0.1

/

test_server_context.py

Copy path

Top

File metadata and controls

• Code
• Blame

126 lines (102 loc) · 5 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

"""Tests for the server-side `Context`.

`Context` extends `BaseContext` (forwarding to a `DispatchContext`) with

`lifespan`, `connection`, and request-scoped `log`. End-to-end tested over

`DirectDispatcher`.

"""

from collections.abc import Mapping

from dataclasses import dataclass

from typing import Any

import anyio

import pytest

from mcp_types import LOG_LEVEL_META_KEY

from mcp_types.version import LATEST_MODERN_VERSION

from mcp.server.connection import Connection

from mcp.server.context import Context

from mcp.shared.dispatcher import DispatchContext

from mcp.shared.transport_context import TransportContext

from ..shared.conftest import direct_pair

from ..shared.test_dispatcher import Recorder, echo_handlers, running_pair

DCtx = DispatchContext[TransportContext]

@dataclass

class _Lifespan:

name: str

@pytest.mark.anyio

async def test_context_exposes_lifespan_and_connection_and_forwards_base_context():

captured: list[Context[_Lifespan]] = []

conn_holder: list[Connection] = []

async def server_on_request(dctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]:

ctx: Context[_Lifespan] = Context(dctx, lifespan=_Lifespan("app"), connection=conn_holder[0])

captured.append(ctx)

return {}

async with running_pair(direct_pair, server_on_request=server_on_request) as (client, server, *_):

conn_holder.append(Connection.for_loop(server, session_id="sess-1"))

with anyio.fail_after(5):

await client.send_raw_request("t", None)

ctx = captured[0]

assert ctx.lifespan.name == "app"

assert ctx.connection is conn_holder[0]

assert ctx.transport.kind == "direct"

assert ctx.can_send_request is True

assert ctx.session_id == "sess-1"

assert ctx.headers is None

@pytest.mark.anyio

async def test_context_log_sends_request_scoped_message_notification():

crec = Recorder()

_, c_notify = echo_handlers(crec)

async def server_on_request(dctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]:

ctx: Context[_Lifespan] = Context(dctx, lifespan=_Lifespan("app"), connection=Connection.for_loop(dctx))

await ctx.log("debug", "hello") # pyright: ignore[reportDeprecated]

return {}

async with running_pair(direct_pair, server_on_request=server_on_request, client_on_notify=c_notify) as (

client,

*_,

):

with anyio.fail_after(5):

await client.send_raw_request("t", None)

await crec.notified.wait()

method, params = crec.notifications[0]

assert method == "notifications/message"

assert params is not None and params["level"] == "debug" and params["data"] == "hello"

@pytest.mark.anyio

async def test_context_log_is_gated_by_the_request_log_level_at_2026():

"""On a 2026 connection an un-opted request delivers nothing; opting in at

`warning` delivers `warning`+ and drops what falls below."""

crec = Recorder()

_, c_notify = echo_handlers(crec)

async def server_on_request(dctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]:

modern = Connection.from_envelope(LATEST_MODERN_VERSION, None, None, outbound=dctx)

silent: Context[_Lifespan] = Context(dctx, lifespan=_Lifespan("app"), connection=modern)

await silent.log("emergency", "dropped: no opt-in") # pyright: ignore[reportDeprecated]

opted: Context[_Lifespan] = Context(

dctx, lifespan=_Lifespan("app"), connection=modern, meta={LOG_LEVEL_META_KEY: "warning"}

)

await opted.log("info", "dropped: below level") # pyright: ignore[reportDeprecated]

await opted.log("warning", "delivered") # pyright: ignore[reportDeprecated]

return {}

async with running_pair(direct_pair, server_on_request=server_on_request, client_on_notify=c_notify) as (

client,

*_,

):

with anyio.fail_after(5):

await client.send_raw_request("t", None)

await crec.notified.wait()

assert [p["data"] for _, p in crec.notifications if p is not None] == ["delivered"]

@pytest.mark.anyio

async def test_context_log_includes_logger_and_meta_when_supplied():

crec = Recorder()

_, c_notify = echo_handlers(crec)

async def server_on_request(dctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]:

ctx: Context[_Lifespan] = Context(dctx, lifespan=_Lifespan("app"), connection=Connection.for_loop(dctx))

await ctx.log("info", "x", logger="my.log", meta={"traceId": "t"}) # pyright: ignore[reportDeprecated]

return {}

async with running_pair(direct_pair, server_on_request=server_on_request, client_on_notify=c_notify) as (

client,

*_,

):

with anyio.fail_after(5):

await client.send_raw_request("t", None)

await crec.notified.wait()

_, params = crec.notifications[0]

assert params is not None

assert params["logger"] == "my.log"

assert params["_meta"] == {"traceId": "t"}