Add a simple module documentation to help LLMs with the pattern

LLMs have issue infering the actual usage pattern from the code
itself. We add a simple module documentation to help with that.
This commit is contained in:
David Soria Parra
2024-11-25 13:43:50 +00:00
parent 54952ca46a
commit 33f0e590c8
4 changed files with 161 additions and 3 deletions

View File

@@ -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 contextvars
import logging import logging
import warnings import warnings

View File

@@ -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 enum import Enum
from typing import Any from typing import Any
@@ -72,8 +111,10 @@ class ServerSession(
return False return False
# Check each experimental capability # Check each experimental capability
for exp_key, exp_value in capability.experimental.items(): for exp_key, exp_value in capability.experimental.items():
if (exp_key not in client_caps.experimental or if (
client_caps.experimental[exp_key] != exp_value): exp_key not in client_caps.experimental
or client_caps.experimental[exp_key] != exp_value
):
return False return False
return True return True
@@ -117,7 +158,6 @@ class ServerSession(
"Received notification before initialization was complete" "Received notification before initialization was complete"
) )
async def send_log_message( async def send_log_message(
self, level: types.LoggingLevel, data: Any, logger: str | None = None self, level: types.LoggingLevel, data: Any, logger: str | None = None
) -> None: ) -> None:

View File

@@ -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 import logging
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from typing import Any from typing import Any

View File

@@ -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 import sys
from contextlib import asynccontextmanager from contextlib import asynccontextmanager