Allow passing initialization options to a session

We need a way for servers to pass initialization options to the session.
This is the beginning of this.
This commit is contained in:
David Soria Parra 2024-10-11 11:06:02 +01:00
commit cc342a05ad
No known key found for this signature in database

View file

@ -32,6 +32,7 @@ from mcp_python.types import (
ReadResourceResult,
Resource,
ResourceReference,
ServerCapabilities,
ServerResult,
SetLevelRequest,
SubscribeRequest,
@ -40,7 +41,6 @@ from mcp_python.types import (
logger = logging.getLogger(__name__)
request_ctx: contextvars.ContextVar[RequestContext] = contextvars.ContextVar(
"request_ctx"
)
@ -53,6 +53,33 @@ class Server:
self.notification_handlers: dict[type, Callable[..., Awaitable[None]]] = {}
logger.info(f"Initializing server '{name}'")
def create_initialization_options(self) -> types.InitializationOptions:
"""Create initialization options from this server instance."""
def pkg_version(package: str) -> str:
try:
from importlib.metadata import version
return version(package)
except Exception:
return "unknown"
return types.InitializationOptions(
server_name=self.name,
server_version=pkg_version("mcp_python"),
capabilities=self.get_capabilities(),
)
def get_capabilities(self) -> ServerCapabilities:
"""Convert existing handlers to a ServerCapabilities object."""
def get_capability(req_type: type) -> dict[str, Any] | None:
return {} if req_type in self.request_handlers else None
return ServerCapabilities(
prompts=get_capability(ListPromptsRequest),
resources=get_capability(ListResourcesRequest),
tools=get_capability(ListPromptsRequest),
logging=get_capability(SetLevelRequest)
)
@property
def request_context(self) -> RequestContext:
"""If called outside of a request context, this will raise a LookupError."""
@ -280,9 +307,10 @@ class Server:
self,
read_stream: MemoryObjectReceiveStream[JSONRPCMessage | Exception],
write_stream: MemoryObjectSendStream[JSONRPCMessage],
initialization_options: types.InitializationOptions
):
with warnings.catch_warnings(record=True) as w:
async with ServerSession(read_stream, write_stream) as session:
async with ServerSession(read_stream, write_stream, initialization_options) as session:
async for message in session.incoming_messages:
logger.debug(f"Received message: {message}")