Skip to content

Add a simple module documentation to help LLMs with the pattern #65

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Nov 25, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 62 additions & 0 deletions src/mcp/server/__init__.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,65 @@
"""
MCP Server Module

This module provides a framework for creating an MCP (Model Context Protocol) server.
It allows you to easily define and handle various types of requests and notifications
in an asynchronous manner.

Usage:
1. Create a Server instance:
server = Server("your_server_name")

2. Define request handlers using decorators:
@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
# Implementation

@server.get_prompt()
async def handle_get_prompt(
name: str, arguments: dict[str, str] | None
) -> types.GetPromptResult:
# Implementation

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
# Implementation

@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
# Implementation

3. Define notification handlers if needed:
@server.progress_notification()
async def handle_progress(
progress_token: str | int, progress: float, total: float | None
) -> None:
# Implementation

4. Run the server:
async def main():
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="your_server_name",
server_version="your_version",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)

asyncio.run(main())

The Server class provides methods to register handlers for various MCP requests and
notifications. It automatically manages the request context and handles incoming
messages from the client.
"""

import contextvars
import logging
import warnings
Expand Down
46 changes: 43 additions & 3 deletions src/mcp/server/session.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,42 @@
"""
ServerSession Module

This module provides the ServerSession class, which manages communication between the
server and client in the MCP (Model Context Protocol) framework. It is most commonly
used in MCP servers to interact with the client.

Common usage pattern:
```
server = Server(name)

@server.call_tool()
async def handle_tool_call(ctx: RequestContext, arguments: dict[str, Any]) -> Any:
# Check client capabilities before proceeding
if ctx.session.check_client_capability(
types.ClientCapabilities(experimental={"advanced_tools": True})
):
# Perform advanced tool operations
result = await perform_advanced_tool_operation(arguments)
else:
# Fall back to basic tool operations
result = await perform_basic_tool_operation(arguments)

return result

@server.list_prompts()
async def handle_list_prompts(ctx: RequestContext) -> list[types.Prompt]:
# Access session for any necessary checks or operations
if ctx.session.client_params:
# Customize prompts based on client initialization parameters
return generate_custom_prompts(ctx.session.client_params)
else:
return default_prompts
```

The ServerSession class is typically used internally by the Server class and should not
be instantiated directly by users of the MCP framework.
"""

from enum import Enum
from typing import Any

Expand Down Expand Up @@ -72,8 +111,10 @@ def check_client_capability(self, capability: types.ClientCapabilities) -> bool:
return False
# Check each experimental capability
for exp_key, exp_value in capability.experimental.items():
if (exp_key not in client_caps.experimental or
client_caps.experimental[exp_key] != exp_value):
if (
exp_key not in client_caps.experimental
or client_caps.experimental[exp_key] != exp_value
):
return False

return True
Expand Down Expand Up @@ -117,7 +158,6 @@ async def _received_notification(
"Received notification before initialization was complete"
)


async def send_log_message(
self, level: types.LoggingLevel, data: Any, logger: str | None = None
) -> None:
Expand Down
36 changes: 36 additions & 0 deletions src/mcp/server/sse.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,39 @@
"""
SSE Server Transport Module

This module implements a Server-Sent Events (SSE) transport layer for MCP servers.

Example usage:
```
# Create an SSE transport at an endpoint
sse = SseServerTransport("/messages")

# Create Starlette routes for SSE and message handling
routes = [
Route("/sse", endpoint=handle_sse),
Route("/messages", endpoint=handle_messages, methods=["POST"])
]

# Define handler functions
async def handle_sse(request):
async with sse.connect_sse(
request.scope, request.receive, request._send
) as streams:
await app.run(
streams[0], streams[1], app.create_initialization_options()
)

async def handle_messages(request):
await sse.handle_post_message(request.scope, request.receive, request._send)

# Create and run Starlette app
starlette_app = Starlette(routes=routes)
uvicorn.run(starlette_app, host="0.0.0.0", port=port)
```

See SseServerTransport class documentation for more details.
"""

import logging
from contextlib import asynccontextmanager
from typing import Any
Expand Down
20 changes: 20 additions & 0 deletions src/mcp/server/stdio.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,23 @@
"""
Stdio Server Transport Module

This module provides functionality for creating an stdio-based transport layer
that can be used to communicate with an MCP client through standard input/output
streams.

Example usage:
```
async def run_server():
async with stdio_server() as (read_stream, write_stream):
# read_stream contains incoming JSONRPCMessages from stdin
# write_stream allows sending JSONRPCMessages to stdout
server = await create_my_server()
await server.run(read_stream, write_stream, init_options)

anyio.run(run_server)
```
"""

import sys
from contextlib import asynccontextmanager

Expand Down