Qiniu MCP Server

import logging from abc import abstractmethod from typing import Dict, AsyncGenerator, Iterable from mcp import types from mcp.server.lowlevel.helper_types import ReadResourceContents from ..consts import consts logger = logging.getLogger(consts.LOGGER_NAME) ResourceContents = str | bytes | Iterable[ReadResourceContents] class ResourceProvider: def __init__(self, scheme: str): self.scheme = scheme @abstractmethod async def list_resources(self, **kwargs) -> list[types.Resource]: pass @abstractmethod async def read_resource(self, uri: types.AnyUrl, **kwargs) -> ResourceContents: pass _all_resource_providers: Dict[str, ResourceProvider] = {} async def list_resources(**kwargs) -> AsyncGenerator[types.Resource, None]: if len(_all_resource_providers) == 0: return for provider in _all_resource_providers.values(): resources = await provider.list_resources(**kwargs) for resource in resources: yield resource return async def read_resource(uri: types.AnyUrl, **kwargs) -> ResourceContents: if len(_all_resource_providers) == 0: return "" provider = _all_resource_providers.get(uri.scheme) return await provider.read_resource(uri=uri, **kwargs) def register_resource_provider(provider: ResourceProvider): """注册工具，禁止重复名称""" name = provider.scheme if name in _all_resource_providers: raise ValueError(f"Resource Provider {name} already registered") _all_resource_providers[name] = provider __all__ = [ "ResourceContents", "ResourceProvider", "list_resources", "read_resource", "register_resource_provider", ]