feat: Add Miku MCP server with S3-style API support

Implemented new Miku MCP server module with the following features:
- Bearer token authentication using QINIU_API_KEY env variable
- S3-style API support for bucket operations
- create_bucket API endpoint (PUT https://<bucket>.<endpoint>)
- Integration with existing MCP server architecture
- Added aiohttp dependency for async HTTP requests

The implementation follows the same pattern as existing storage/cdn/media_processing modules.

Generated with [codeagent](https://github.com/qbox/codeagent)
Co-authored-by: callmefisher <[email protected]>

Author: xgopilot

pyproject.toml

@@ -8,6 +8,7 @@ authors = [
 ]
 keywords = ["qiniu", "mcp", "llm"]
 dependencies = [
+    "aiohttp>=3.9.0",
     "aioboto3>=13.2.0",
     "fastjsonschema>=2.21.1",
     "httpx>=0.28.1",

src/mcp_server/core/__init__.py

@@ -3,6 +3,7 @@
 from .media_processing import load as load_media_processing
 from .cdn import load as load_cdn
 from .version import load as load_version
+from .miku import load as load_miku
 
 
 def load():
@@ -17,4 +18,6 @@ def load():
     load_cdn(cfg)
     # 智能多媒体
     load_media_processing(cfg)
+    # Miku
+    load_miku(cfg)
 

src/mcp_server/core/miku/__init__.py

@@ -0,0 +1,11 @@
+from .miku import MikuService
+from .tools import register_tools
+from ...config import config
+
+
+def load(cfg: config.Config):
+    miku = MikuService(cfg)
+    register_tools(miku)
+
+
+__all__ = ["load"]

src/mcp_server/core/miku/miku.py

@@ -0,0 +1,77 @@
+import aiohttp
+import logging
+
+from typing import Dict, Any
+from ...config import config
+from ...consts import consts
+
+logger = logging.getLogger(consts.LOGGER_NAME)
+
+
+class MikuService:
+    def __init__(self, cfg: config.Config = None):
+        self.config = cfg
+        self.api_key = cfg.access_key if cfg else None
+        self.endpoint_url = cfg.endpoint_url if cfg else None
+
+    def _get_auth_header(self) -> Dict[str, str]:
+        """Generate Bearer token authorization header"""
+        if not self.api_key:
+            raise ValueError("QINIU_API_KEY is not configured")
+        return {
+            "Authorization": f"Bearer {self.api_key}"
+        }
+
+    def _build_bucket_url(self, bucket: str) -> str:
+        """Build S3-style bucket URL"""
+        if not self.endpoint_url:
+            raise ValueError("QINIU_ENDPOINT_URL is not configured")
+
+        # Remove protocol if present in endpoint_url
+        endpoint = self.endpoint_url
+        if endpoint.startswith("http://"):
+            endpoint = endpoint[7:]
+        elif endpoint.startswith("https://"):
+            endpoint = endpoint[8:]
+
+        # Build URL in format: https://<bucket>.<endpoint>
+        return f"https://{bucket}.{endpoint}"
+
+    async def create_bucket(self, bucket: str) -> Dict[str, Any]:
+        """
+        Create a bucket using S3-style API
+
+        Args:
+            bucket: The bucket name to create
+
+        Returns:
+            Dict containing the response status and message
+        """
+        url = self._build_bucket_url(bucket)
+        headers = self._get_auth_header()
+
+        logger.info(f"Creating bucket: {bucket} at {url}")
+
+        async with aiohttp.ClientSession() as session:
+            async with session.put(url, headers=headers) as response:
+                status = response.status
+                text = await response.text()
+
+                if status == 200 or status == 201:
+                    logger.info(f"Successfully created bucket: {bucket}")
+                    return {
+                        "status": "success",
+                        "bucket": bucket,
+                        "url": url,
+                        "message": f"Bucket '{bucket}' created successfully",
+                        "status_code": status
+                    }
+                else:
+                    logger.error(f"Failed to create bucket: {bucket}, status: {status}, response: {text}")
+                    return {
+                        "status": "error",
+                        "bucket": bucket,
+                        "url": url,
+                        "message": f"Failed to create bucket: {text}",
+                        "status_code": status
+                    }

src/mcp_server/core/miku/tools.py

@@ -0,0 +1,46 @@
+import logging
+
+from mcp import types
+from mcp.types import TextContent
+
+from .miku import MikuService
+from ...consts import consts
+from ...tools import tools
+
+logger = logging.getLogger(consts.LOGGER_NAME)
+
+_BUCKET_DESC = "Miku bucket name"
+
+
+class _ToolImpl:
+    def __init__(self, miku: MikuService):
+        self.miku = miku
+
+    @tools.tool_meta(
+        types.Tool(
+            name="miku_create_bucket",
+            description="Create a new bucket in Miku using S3-style API. The bucket will be created at https://<bucket>.<endpoint_url>",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "bucket": {
+                        "type": "string",
+                        "description": _BUCKET_DESC,
+                    },
+                },
+                "required": ["bucket"],
+            },
+        )
+    )
+    async def create_bucket(self, **kwargs) -> list[types.TextContent]:
+        result = await self.miku.create_bucket(**kwargs)
+        return [types.TextContent(type="text", text=str(result))]
+
+
+def register_tools(miku: MikuService):
+    tool_impl = _ToolImpl(miku)
+    tools.auto_register_tools(
+        [
+            tool_impl.create_bucket,
+        ]
+    )