0
# Low-Level Server Framework
1

2
Low-level server implementation providing full control over MCP protocol handling. The Server class offers decorator-based request handlers, custom lifecycle management, and fine-grained control over protocol capabilities and session management.
3

4
## Capabilities
5

6
### Server Class
7

8
Core server class for implementing MCP protocol handlers with full control over request processing and capabilities.
9

10
```python { .api }
11
class Server:
12
    def __init__(
13
        self,
14
        name: str,
15
        version: str | None = None,
16
        instructions: str | None = None,
17
        lifespan: Callable = default_lifespan,
18
    ):
19
        """
20
        Initialize a low-level MCP server.
21

22
        Parameters:
23
        - name: Server name for identification
24
        - version: Server version string
25
        - instructions: Server instructions/description
26
        - lifespan: Server lifespan management function
27
        """
28

29
    def list_tools(self) -> Callable:
30
        """
31
        Decorator for tool listing handler.
32

33
        Returns:
34
        Decorator function that expects handler returning list[Tool]
35
        """
36

37
    def call_tool(self) -> Callable:
38
        """
39
        Decorator for tool execution handler.
40

41
        Returns:
42
        Decorator function that expects handler with (name: str, arguments: dict) -> list[ContentBlock]
43
        """
44

45
    def list_resources(self) -> Callable:
46
        """
47
        Decorator for resource listing handler.
48

49
        Returns:
50
        Decorator function that expects handler returning list[Resource]
51
        """
52

53
    def read_resource(self) -> Callable:
54
        """
55
        Decorator for resource reading handler.
56

57
        Returns:
58
        Decorator function that expects handler with (uri: AnyUrl) -> list[ContentBlock]
59
        """
60

61
    def subscribe_resource(self) -> Callable:
62
        """
63
        Decorator for resource subscription handler.
64

65
        Returns:
66
        Decorator function that expects handler with (uri: AnyUrl) -> None
67
        """
68

69
    def unsubscribe_resource(self) -> Callable:
70
        """
71
        Decorator for resource unsubscription handler.
72

73
        Returns:
74
        Decorator function that expects handler with (uri: AnyUrl) -> None
75
        """
76

77
    def list_prompts(self) -> Callable:
78
        """
79
        Decorator for prompt listing handler.
80

81
        Returns:
82
        Decorator function that expects handler returning list[Prompt]
83
        """
84

85
    def get_prompt(self) -> Callable:
86
        """
87
        Decorator for prompt retrieval handler.
88

89
        Returns:
90
        Decorator function that expects handler with (name: str, arguments: dict) -> GetPromptResult
91
        """
92

93
    def set_logging_level(self) -> Callable:
94
        """
95
        Decorator for logging level setting handler.
96

97
        Returns:
98
        Decorator function that expects handler with (level: LoggingLevel) -> None
99
        """
100

101
    def completion(self) -> Callable:
102
        """
103
        Decorator for completion handler.
104

105
        Returns:
106
        Decorator function that expects handler with completion request -> list[Completion]
107
        """
108

109
    async def run(
110
        self,
111
        read_stream: MemoryObjectReceiveStream,
112
        write_stream: MemoryObjectSendStream,
113
        options: InitializationOptions
114
    ) -> None:
115
        """
116
        Run the server with provided streams and options.
117

118
        Parameters:
119
        - read_stream: Stream for receiving client messages
120
        - write_stream: Stream for sending responses to client
121
        - options: Server initialization options
122
        """
123

124
    async def create_initialization_options(self) -> InitializationOptions:
125
        """
126
        Create initialization options for the server.
127

128
        Returns:
129
        InitializationOptions with server capabilities
130
        """
131
```
132

133
### Configuration Classes
134

135
Configuration and options classes for server initialization and capabilities.
136

137
```python { .api }
138
class InitializationOptions:
139
    def __init__(
140
        self,
141
        server_name: str,
142
        server_version: str,
143
        capabilities: ServerCapabilities,
144
        instructions: str | None = None,
145
    ):
146
        """
147
        Server initialization options.
148

149
        Parameters:
150
        - server_name: Name of the server
151
        - server_version: Version of the server
152
        - capabilities: Server capabilities declaration
153
        - instructions: Server instructions/description
154
        """
155

156
class NotificationOptions:
157
    def __init__(
158
        self,
159
        tools_changed: bool = True,
160
        resources_changed: bool = True,
161
        prompts_changed: bool = True,
162
        logging: bool = True,
163
        progress: bool = True,
164
    ):
165
        """
166
        Notification capability options.
167

168
        Parameters:
169
        - tools_changed: Enable tool change notifications
170
        - resources_changed: Enable resource change notifications
171
        - prompts_changed: Enable prompt change notifications
172
        - logging: Enable logging notifications
173
        - progress: Enable progress notifications
174
        """
175
```
176

177
### Context and Session Access
178

179
Low-level context access for request handling and session management.
180

181
```python { .api }
182
def request_ctx() -> RequestContext:
183
    """
184
    Get the current request context.
185

186
    Returns:
187
    RequestContext for the current request
188
    """
189

190
class RequestContext:
191
    @property
192
    def request_id(self) -> str:
193
        """Current request identifier."""
194

195
    @property
196
    def session(self) -> ServerSession:
197
        """Current server session."""
198

199
    @property
200
    def client_info(self) -> Implementation:
201
        """Client implementation information."""
202

203
    @property
204
    def server_info(self) -> Implementation:
205
        """Server implementation information."""
206

207
    async def send_notification(
208
        self,
209
        method: str,
210
        params: dict[str, Any] | None = None
211
    ) -> None:
212
        """
213
        Send notification to client.
214

215
        Parameters:
216
        - method: Notification method name
217
        - params: Notification parameters
218
        """
219

220
    async def send_progress_notification(
221
        self,
222
        progress_token: ProgressToken,
223
        progress: float,
224
        total: float | None = None
225
    ) -> None:
226
        """
227
        Send progress notification to client.
228

229
        Parameters:
230
        - progress_token: Progress tracking token
231
        - progress: Current progress value
232
        - total: Total expected value
233
        """
234

235
    async def send_logging_message(
236
        self,
237
        level: LoggingLevel,
238
        message: str,
239
        logger: str | None = None
240
    ) -> None:
241
        """
242
        Send log message to client.
243

244
        Parameters:
245
        - level: Log level
246
        - message: Log message
247
        - logger: Logger name
248
        """
249
```
250

251
## Usage Examples
252

253
### Basic Low-Level Server
254

255
```python
256
from mcp.server import Server
257
from mcp import Tool, Resource, TextContent
258
import asyncio
259

260
# Create server instance
261
server = Server("basic-server", version="1.0.0")
262

263
# Tool management
264
available_tools = [
265
    Tool(
266
        name="echo",
267
        description="Echo the input message",
268
        inputSchema={
269
            "type": "object",
270
            "properties": {
271
                "message": {"type": "string"}
272
            },
273
            "required": ["message"]
274
        }
275
    )
276
]
277

278
@server.list_tools()
279
async def list_tools():
280
    """Return list of available tools."""
281
    return available_tools
282

283
@server.call_tool()
284
async def call_tool(name: str, arguments: dict):
285
    """Handle tool execution."""
286
    if name == "echo":
287
        message = arguments.get("message", "")
288
        return [TextContent(type="text", text=f"Echo: {message}")]
289
    else:
290
        raise ValueError(f"Unknown tool: {name}")
291

292
# Resource management
293
available_resources = [
294
    Resource(
295
        uri="memory://stats",
296
        name="Server Statistics",
297
        description="Current server statistics"
298
    )
299
]
300

301
@server.list_resources()
302
async def list_resources():
303
    """Return list of available resources."""
304
    return available_resources
305

306
@server.read_resource()
307
async def read_resource(uri):
308
    """Handle resource reading."""
309
    if str(uri) == "memory://stats":
310
        stats = {
311
            "uptime": "1 hour",
312
            "requests": 42,
313
            "tools_called": 12
314
        }
315
        content = "\n".join(f"{k}: {v}" for k, v in stats.items())
316
        return [TextContent(type="text", text=content)]
317
    else:
318
        raise ValueError(f"Unknown resource: {uri}")
319

320
# Run server with stdio transport
321
async def main():
322
    from mcp.server import stdio_server
323
    from mcp.server.models import InitializationOptions
324
    from mcp import ServerCapabilities, ToolsCapability, ResourcesCapability
325
    
326
    # Create initialization options
327
    capabilities = ServerCapabilities(
328
        tools=ToolsCapability(listChanged=True),
329
        resources=ResourcesCapability(listChanged=True, subscribe=True)
330
    )
331
    
332
    options = InitializationOptions(
333
        server_name="basic-server",
334
        server_version="1.0.0", 
335
        capabilities=capabilities
336
    )
337
    
338
    async with stdio_server() as (read, write):
339
        await server.run(read, write, options)
340

341
if __name__ == "__main__":
342
    asyncio.run(main())
343
```
344

345
### Server with Custom Capabilities
346

347
```python
348
from mcp.server import Server, request_ctx
349
from mcp import *
350
import asyncio
351

352
server = Server("advanced-server")
353

354
# Custom completion handler
355
@server.completion()
356
async def handle_completion(request):
357
    """Handle completion requests for resources and prompts."""
358
    ctx = request_ctx()
359
    
360
    if request.ref.type == "resource":
361
        # Complete resource URIs
362
        if request.ref.uri.startswith("file://"):
363
            # Return file path completions
364
            return [
365
                Completion(
366
                    values=["file:///tmp/example.txt", "file:///home/user/doc.md"],
367
                    total=2
368
                )
369
            ]
370
    
371
    return [Completion(values=[], total=0)]
372

373
# Custom logging level handler
374
@server.set_logging_level()
375
async def set_logging_level(level):
376
    """Handle logging level changes."""
377
    ctx = request_ctx()
378
    print(f"Client {ctx.client_info.name} set logging level to {level}")
379
    
380
    # Send confirmation
381
    await ctx.send_logging_message(
382
        LoggingLevel.INFO,
383
        f"Logging level set to {level}"
384
    )
385

386
# Prompt with arguments
387
prompts = [
388
    Prompt(
389
        name="code_review",
390
        description="Generate code review checklist",
391
        arguments=[
392
            PromptArgument(
393
                name="language",
394
                description="Programming language",
395
                required=True
396
            ),
397
            PromptArgument(
398
                name="complexity",
399
                description="Code complexity level",
400
                required=False
401
            )
402
        ]
403
    )
404
]
405

406
@server.list_prompts()
407
async def list_prompts():
408
    """Return available prompts."""
409
    return prompts
410

411
@server.get_prompt()
412
async def get_prompt(name: str, arguments: dict):
413
    """Handle prompt generation."""
414
    if name == "code_review":
415
        language = arguments.get("language", "Python")
416
        complexity = arguments.get("complexity", "medium")
417
        
418
        prompt_text = f"""Code Review Checklist for {language} ({complexity} complexity):
419

420
1. Code Style and Formatting
421
   - Consistent indentation and spacing
422
   - Meaningful variable and function names
423
   - Appropriate comments and documentation
424

425
2. Logic and Functionality
426
   - Code achieves intended purpose
427
   - Edge cases are handled
428
   - Error handling is appropriate
429

430
3. Performance Considerations
431
   - Efficient algorithms and data structures
432
   - Resource usage optimization
433
   - Scalability considerations
434

435
4. Security Review
436
   - Input validation
437
   - Authentication and authorization
438
   - Data sanitization
439
"""
440
        
441
        return GetPromptResult(
442
            description=f"Code review checklist for {language}",
443
            messages=[
444
                PromptMessage(
445
                    role=Role.user,
446
                    content=TextContent(type="text", text=prompt_text)
447
                )
448
            ]
449
        )
450
    else:
451
        raise ValueError(f"Unknown prompt: {name}")
452

453
# Resource subscription handling
454
@server.subscribe_resource()
455
async def subscribe_resource(uri):
456
    """Handle resource subscription."""
457
    ctx = request_ctx()
458
    print(f"Client subscribed to resource: {uri}")
459
    
460
    # Send initial notification
461
    await ctx.send_notification(
462
        "notifications/resources/updated",
463
        {"uri": str(uri), "reason": "subscribed"}
464
    )
465

466
@server.unsubscribe_resource()
467
async def unsubscribe_resource(uri):
468
    """Handle resource unsubscription."""
469
    ctx = request_ctx()
470
    print(f"Client unsubscribed from resource: {uri}")
471

472
async def main():
473
    from mcp.server import stdio_server
474
    
475
    options = await server.create_initialization_options()
476
    
477
    async with stdio_server() as (read, write):
478
        await server.run(read, write, options)
479

480
if __name__ == "__main__":
481
    asyncio.run(main())
482
```
483

484
### Server with Progress Tracking
485

486
```python
487
from mcp.server import Server, request_ctx
488
from mcp import TextContent
489
import asyncio
490

491
server = Server("progress-server")
492

493
@server.call_tool()
494
async def call_tool(name: str, arguments: dict):
495
    """Handle tool calls with progress tracking."""
496
    ctx = request_ctx()
497
    
498
    if name == "process_data":
499
        items = arguments.get("items", [])
500
        total = len(items)
501
        
502
        results = []
503
        for i, item in enumerate(items):
504
            # Send progress update
505
            await ctx.send_progress_notification(
506
                progress_token=ctx.request_id,
507
                progress=i,
508
                total=total
509
            )
510
            
511
            # Simulate processing
512
            await asyncio.sleep(0.1)
513
            results.append(f"Processed: {item}")
514
        
515
        # Send final progress
516
        await ctx.send_progress_notification(
517
            progress_token=ctx.request_id,
518
            progress=total,
519
            total=total
520
        )
521
        
522
        return [TextContent(
523
            type="text", 
524
            text=f"Completed processing {total} items: {results}"
525
        )]
526
    
527
    return [TextContent(type="text", text="Unknown tool")]
528

529
@server.list_tools()
530
async def list_tools():
531
    """Return tools that support progress tracking."""
532
    return [
533
        Tool(
534
            name="process_data",
535
            description="Process data items with progress tracking",
536
            inputSchema={
537
                "type": "object",
538
                "properties": {
539
                    "items": {
540
                        "type": "array",
541
                        "items": {"type": "string"}
542
                    }
543
                },
544
                "required": ["items"]
545
            }
546
        )
547
    ]
548

549
async def main():
550
    from mcp.server import stdio_server
551
    
552
    options = await server.create_initialization_options()
553
    
554
    async with stdio_server() as (read, write):
555
        await server.run(read, write, options)
556

557
if __name__ == "__main__":
558
    asyncio.run(main())
559
```
560

561
### Custom Lifespan Management
562

563
```python
564
from mcp.server import Server
565
import asyncio
566
import aiofiles
567

568
# Custom lifespan function
569
async def custom_lifespan(server_instance):
570
    """Custom server lifespan with setup and cleanup."""
571
    print("Server starting up...")
572
    
573
    # Setup: Initialize resources, connections, etc.
574
    server_instance._data_file = await aiofiles.open("server_data.json", "w")
575
    server_instance._active_connections = set()
576
    
577
    try:
578
        # Yield control to the server
579
        yield
580
    finally:
581
        # Cleanup: Close resources, save state, etc.
582
        print("Server shutting down...")
583
        await server_instance._data_file.close()
584
        print(f"Closed {len(server_instance._active_connections)} connections")
585

586
# Create server with custom lifespan
587
server = Server(
588
    "lifecycle-server",
589
    version="1.0.0",
590
    lifespan=custom_lifespan
591
)
592

593
@server.list_tools()
594
async def list_tools():
595
    """Return simple tool list."""
596
    return [
597
        Tool(
598
            name="status",
599
            description="Get server status",
600
            inputSchema={"type": "object", "properties": {}}
601
        )
602
    ]
603

604
@server.call_tool()
605
async def call_tool(name: str, arguments: dict):
606
    """Handle tool calls with server state access."""
607
    if name == "status":
608
        connections = len(getattr(server, '_active_connections', []))
609
        return [TextContent(
610
            type="text",
611
            text=f"Server is running with {connections} active connections"
612
        )]
613
    
614
    return [TextContent(type="text", text="Unknown tool")]
615

616
async def main():
617
    from mcp.server import stdio_server
618
    
619
    options = await server.create_initialization_options()
620
    
621
    async with stdio_server() as (read, write):
622
        await server.run(read, write, options)
623

624
if __name__ == "__main__":
625
    asyncio.run(main())
626
```