feat: optimize miku MCP server - remove unused import and add stream creation API

1. Remove unused import mcp.types.TextContent from tools.py
2. Add create_stream API implementation with S3-style format: https://<bucket>.<endpoint>/<stream>
3. Add miku_create_stream MCP tool for stream creation

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

Author: xgopilot

src/mcp_server/core/miku/miku.py

@@ -37,6 +37,21 @@ def _build_bucket_url(self, bucket: str) -> str:
         # Build URL in format: https://<bucket>.<endpoint>
         return f"https://{bucket}.{endpoint}"
 
+    def _build_stream_url(self, bucket: str, stream: str) -> str:
+        """Build S3-style stream 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>/<stream>
+        return f"https://{bucket}.{endpoint}/{stream}"
+
     async def create_bucket(self, bucket: str) -> Dict[str, Any]:
         """
         Create a bucket using S3-style API
@@ -75,3 +90,45 @@ async def create_bucket(self, bucket: str) -> Dict[str, Any]:
                         "message": f"Failed to create bucket: {text}",
                         "status_code": status
                     }
+
+    async def create_stream(self, bucket: str, stream: str) -> Dict[str, Any]:
+        """
+        Create a stream using S3-style API
+
+        Args:
+            bucket: The bucket name
+            stream: The stream name to create
+
+        Returns:
+            Dict containing the response status and message
+        """
+        url = self._build_stream_url(bucket, stream)
+        headers = self._get_auth_header()
+
+        logger.info(f"Creating stream: {stream} in 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 stream: {stream} in bucket: {bucket}")
+                    return {
+                        "status": "success",
+                        "bucket": bucket,
+                        "stream": stream,
+                        "url": url,
+                        "message": f"Stream '{stream}' created successfully in bucket '{bucket}'",
+                        "status_code": status
+                    }
+                else:
+                    logger.error(f"Failed to create stream: {stream}, status: {status}, response: {text}")
+                    return {
+                        "status": "error",
+                        "bucket": bucket,
+                        "stream": stream,
+                        "url": url,
+                        "message": f"Failed to create stream: {text}",
+                        "status_code": status
+                    }

src/mcp_server/core/miku/tools.py

@@ -1,7 +1,6 @@
 import logging
 
 from mcp import types
-from mcp.types import TextContent
 
 from .miku import MikuService
 from ...consts import consts
@@ -10,6 +9,7 @@
 logger = logging.getLogger(consts.LOGGER_NAME)
 
 _BUCKET_DESC = "Miku bucket name"
+_STREAM_DESC = "Miku stream name"
 
 
 class _ToolImpl:
@@ -36,11 +36,36 @@ async def create_bucket(self, **kwargs) -> list[types.TextContent]:
         result = await self.miku.create_bucket(**kwargs)
         return [types.TextContent(type="text", text=str(result))]
 
+    @tools.tool_meta(
+        types.Tool(
+            name="miku_create_stream",
+            description="Create a new stream in Miku using S3-style API. The stream will be created at https://<bucket>.<endpoint_url>/<stream>",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "bucket": {
+                        "type": "string",
+                        "description": _BUCKET_DESC,
+                    },
+                    "stream": {
+                        "type": "string",
+                        "description": _STREAM_DESC,
+                    },
+                },
+                "required": ["bucket", "stream"],
+            },
+        )
+    )
+    async def create_stream(self, **kwargs) -> list[types.TextContent]:
+        result = await self.miku.create_stream(**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,
+            tool_impl.create_stream,
         ]
     )