Qiniu MCP Server

Instructions

Implementation Reference

• src/mcp_server/core/storage/tools.py:91-109 (handler)

  The MCP tool handler for 'get_object', which retrieves the object from the StorageService, processes it based on content type (image or text), and returns appropriate MCP content types (ImageContent or TextContent).

  async def get_object(self, **kwargs) -> list[ImageContent] | list[TextContent]:
      response = await self.storage.get_object(**kwargs)
      file_content = response["Body"]
      content_type = response.get("ContentType", "application/octet-stream")
  
      # 根据内容类型返回不同的响应
      if content_type.startswith("image/"):
          base64_data = base64.b64encode(file_content).decode("utf-8")
          return [
              types.ImageContent(
                  type="image", data=base64_data, mimeType=content_type
              )
          ]
  
      if isinstance(file_content, bytes):
          text_content = file_content.decode("utf-8")
      else:
          text_content = str(file_content)
      return [types.TextContent(type="text", text=text_content)]

• src/mcp_server/core/storage/tools.py:72-90 (schema)

  The input schema and metadata definition for the 'get_object' tool, including name, description, and inputSchema requiring bucket and key.

      types.Tool(
          name="get_object",
          description="Get an object contents from Qiniu Cloud bucket. In the GetObject request, specify the full key name for the object.",
          inputSchema={
              "type": "object",
              "properties": {
                  "bucket": {
                      "type": "string",
                      "description": _BUCKET_DESC,
                  },
                  "key": {
                      "type": "string",
                      "description": "Key of the object to get.",
                  },
              },
              "required": ["bucket", "key"],
          },
      )
  )

• src/mcp_server/core/storage/tools.py:236-248 (registration)

  The register_tools function that instantiates _ToolImpl with StorageService and registers the tool handlers, including get_object, via tools.auto_register_tools.

  def register_tools(storage: StorageService):
      tool_impl = _ToolImpl(storage)
      tools.auto_register_tools(
          [
              tool_impl.list_buckets,
              tool_impl.list_objects,
              tool_impl.get_object,
              tool_impl.upload_text_data,
              tool_impl.upload_local_file,
              tool_impl.get_object_url,
          ]
      )

• src/mcp_server/core/storage/__init__.py:7-11 (registration)

  The load function in storage/__init__.py that creates StorageService and calls register_tools to register the tools including 'get_object'.

  def load(cfg: config.Config):
      storage = StorageService(cfg)
      register_tools(storage)
      register_resource_provider(storage)

• src/mcp_server/core/storage/storage.py:137-162 (helper)

  The underlying StorageService.get_object method called by the tool handler, which performs the S3-compatible get_object operation, reads the full body into bytes, and returns the response dict.

  async def get_object(self, bucket: str, key: str) -> Dict[str, Any]:
      if self.config.buckets and bucket not in self.config.buckets:
          logger.warning(f"Bucket {bucket} not in configured bucket list")
          return {}
  
      async with self.s3_session.client(
              "s3",
              aws_access_key_id=self.config.access_key,
              aws_secret_access_key=self.config.secret_key,
              endpoint_url=self.config.endpoint_url,
              region_name=self.config.region_name,
              config=self.s3_config,
      ) as s3:
          # Get the object and its stream
          response = await s3.get_object(Bucket=bucket, Key=key)
          stream = response["Body"]
  
          # Read the entire stream in chunks
          chunks = []
          async for chunk in stream:
              chunks.append(chunk)
  
          # Replace the stream with the complete data
          response["Body"] = b"".join(chunks)
          return response