Fix MiniMax MCP startup handling
This commit is contained in:
@@ -1,6 +1,8 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
|
from collections import deque
|
||||||
|
from contextlib import suppress
|
||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
@@ -107,6 +109,9 @@ class MinimaxImageUnderstandClient:
|
|||||||
self._request_id = 0
|
self._request_id = 0
|
||||||
self._stdout: asyncio.StreamReader | None = None
|
self._stdout: asyncio.StreamReader | None = None
|
||||||
self._stdin: asyncio.StreamWriter | None = None
|
self._stdin: asyncio.StreamWriter | None = None
|
||||||
|
self._stderr: asyncio.StreamReader | None = None
|
||||||
|
self._stderr_task: asyncio.Task[None] | None = None
|
||||||
|
self._stderr_tail: deque[str] = deque(maxlen=20)
|
||||||
self._lock = asyncio.Lock()
|
self._lock = asyncio.Lock()
|
||||||
|
|
||||||
async def __aenter__(self) -> "MinimaxImageUnderstandClient":
|
async def __aenter__(self) -> "MinimaxImageUnderstandClient":
|
||||||
@@ -129,7 +134,7 @@ class MinimaxImageUnderstandClient:
|
|||||||
*self.config.command,
|
*self.config.command,
|
||||||
stdin=asyncio.subprocess.PIPE,
|
stdin=asyncio.subprocess.PIPE,
|
||||||
stdout=asyncio.subprocess.PIPE,
|
stdout=asyncio.subprocess.PIPE,
|
||||||
stderr=asyncio.subprocess.DEVNULL,
|
stderr=asyncio.subprocess.PIPE,
|
||||||
env=env,
|
env=env,
|
||||||
)
|
)
|
||||||
except FileNotFoundError as exc:
|
except FileNotFoundError as exc:
|
||||||
@@ -138,25 +143,32 @@ class MinimaxImageUnderstandClient:
|
|||||||
) from exc
|
) from exc
|
||||||
|
|
||||||
if self._process.stdin is None or self._process.stdout is None:
|
if self._process.stdin is None or self._process.stdout is None:
|
||||||
|
await self._terminate_process()
|
||||||
raise MCPError("MiniMax MCP process did not expose stdio pipes")
|
raise MCPError("MiniMax MCP process did not expose stdio pipes")
|
||||||
|
|
||||||
self._stdin = self._process.stdin
|
self._stdin = self._process.stdin
|
||||||
self._stdout = self._process.stdout
|
self._stdout = self._process.stdout
|
||||||
|
self._stderr = self._process.stderr
|
||||||
|
self._stderr_task = asyncio.create_task(self._drain_stderr())
|
||||||
request_timeout = self.config.timeout if self.config.timeout is not None else 120.0
|
request_timeout = self.config.timeout if self.config.timeout is not None else 120.0
|
||||||
|
|
||||||
await self._request(
|
try:
|
||||||
"initialize",
|
await self._request(
|
||||||
{
|
"initialize",
|
||||||
"protocolVersion": "2024-11-05",
|
{
|
||||||
"capabilities": {},
|
"protocolVersion": "2024-11-05",
|
||||||
"clientInfo": {
|
"capabilities": {},
|
||||||
"name": "multi-bot-tg",
|
"clientInfo": {
|
||||||
"version": "1.0.0",
|
"name": "multi-bot-tg",
|
||||||
|
"version": "1.0.0",
|
||||||
|
},
|
||||||
},
|
},
|
||||||
},
|
timeout=request_timeout,
|
||||||
timeout=request_timeout,
|
)
|
||||||
)
|
await self._notify("notifications/initialized", {})
|
||||||
await self._notify("notifications/initialized", {})
|
except Exception:
|
||||||
|
await self._terminate_process()
|
||||||
|
raise
|
||||||
|
|
||||||
async def close(self) -> None:
|
async def close(self) -> None:
|
||||||
process = self._process
|
process = self._process
|
||||||
@@ -186,6 +198,8 @@ class MinimaxImageUnderstandClient:
|
|||||||
|
|
||||||
self._stdin = None
|
self._stdin = None
|
||||||
self._stdout = None
|
self._stdout = None
|
||||||
|
self._stderr = None
|
||||||
|
await self._stop_stderr_task()
|
||||||
|
|
||||||
async def list_tools(self) -> list[dict]:
|
async def list_tools(self) -> list[dict]:
|
||||||
if self._process is None:
|
if self._process is None:
|
||||||
@@ -274,7 +288,13 @@ class MinimaxImageUnderstandClient:
|
|||||||
|
|
||||||
if timeout is None:
|
if timeout is None:
|
||||||
return await _wait_for_response()
|
return await _wait_for_response()
|
||||||
return await asyncio.wait_for(_wait_for_response(), timeout=timeout)
|
try:
|
||||||
|
return await asyncio.wait_for(_wait_for_response(), timeout=timeout)
|
||||||
|
except TimeoutError as exc:
|
||||||
|
raise MCPError(
|
||||||
|
f"Timed out waiting for MiniMax MCP '{method}' response after {timeout:.0f}s"
|
||||||
|
f"{self._format_stderr_tail()}"
|
||||||
|
) from exc
|
||||||
|
|
||||||
async def _write_message(self, payload: dict[str, Any]) -> None:
|
async def _write_message(self, payload: dict[str, Any]) -> None:
|
||||||
if self._stdin is None:
|
if self._stdin is None:
|
||||||
@@ -289,10 +309,11 @@ class MinimaxImageUnderstandClient:
|
|||||||
if self._stdout is None:
|
if self._stdout is None:
|
||||||
raise MCPError("MiniMax MCP stdout is not available")
|
raise MCPError("MiniMax MCP stdout is not available")
|
||||||
|
|
||||||
header_block = await self._stdout.readuntil(b"\r\n\r\n")
|
header_block = await self._read_header_block()
|
||||||
headers = header_block.decode("ascii", errors="ignore").split("\r\n")
|
headers = header_block.decode("ascii", errors="ignore").split("\r\n")
|
||||||
content_length = None
|
content_length = None
|
||||||
for header in headers:
|
for header in headers:
|
||||||
|
header = header.strip()
|
||||||
if header.lower().startswith("content-length:"):
|
if header.lower().startswith("content-length:"):
|
||||||
content_length = int(header.split(":", 1)[1].strip())
|
content_length = int(header.split(":", 1)[1].strip())
|
||||||
break
|
break
|
||||||
@@ -302,6 +323,70 @@ class MinimaxImageUnderstandClient:
|
|||||||
payload = await self._stdout.readexactly(content_length)
|
payload = await self._stdout.readexactly(content_length)
|
||||||
return json.loads(payload.decode("utf-8"))
|
return json.loads(payload.decode("utf-8"))
|
||||||
|
|
||||||
|
async def _read_header_block(self) -> bytes:
|
||||||
|
if self._stdout is None:
|
||||||
|
raise MCPError("MiniMax MCP stdout is not available")
|
||||||
|
|
||||||
|
buffer = bytearray()
|
||||||
|
while True:
|
||||||
|
chunk = await self._stdout.read(1)
|
||||||
|
if not chunk:
|
||||||
|
raise MCPError(f"MiniMax MCP stdout closed before a response{self._format_stderr_tail()}")
|
||||||
|
buffer.extend(chunk)
|
||||||
|
if buffer.endswith(b"\r\n\r\n"):
|
||||||
|
return bytes(buffer)
|
||||||
|
if buffer.endswith(b"\n\n"):
|
||||||
|
return bytes(buffer).replace(b"\n", b"\r\n")
|
||||||
|
if len(buffer) > 64 * 1024:
|
||||||
|
raise MCPError("MiniMax MCP response headers exceeded 64KB")
|
||||||
|
|
||||||
|
async def _drain_stderr(self) -> None:
|
||||||
|
if self._stderr is None:
|
||||||
|
return
|
||||||
|
while True:
|
||||||
|
line = await self._stderr.readline()
|
||||||
|
if not line:
|
||||||
|
return
|
||||||
|
text = line.decode("utf-8", errors="replace").strip()
|
||||||
|
if text:
|
||||||
|
self._stderr_tail.append(text)
|
||||||
|
|
||||||
|
def _format_stderr_tail(self) -> str:
|
||||||
|
if not self._stderr_tail:
|
||||||
|
return ""
|
||||||
|
return "\nMiniMax MCP stderr tail:\n" + "\n".join(self._stderr_tail)
|
||||||
|
|
||||||
|
async def _stop_stderr_task(self) -> None:
|
||||||
|
task = self._stderr_task
|
||||||
|
self._stderr_task = None
|
||||||
|
if task is None:
|
||||||
|
return
|
||||||
|
task.cancel()
|
||||||
|
with suppress(asyncio.CancelledError):
|
||||||
|
await task
|
||||||
|
|
||||||
|
async def _terminate_process(self) -> None:
|
||||||
|
process = self._process
|
||||||
|
self._process = None
|
||||||
|
if process is None:
|
||||||
|
return
|
||||||
|
|
||||||
|
try:
|
||||||
|
process.terminate()
|
||||||
|
except ProcessLookupError:
|
||||||
|
pass
|
||||||
|
try:
|
||||||
|
await asyncio.wait_for(process.wait(), timeout=5)
|
||||||
|
except Exception:
|
||||||
|
try:
|
||||||
|
process.kill()
|
||||||
|
except ProcessLookupError:
|
||||||
|
pass
|
||||||
|
await self._stop_stderr_task()
|
||||||
|
self._stdin = None
|
||||||
|
self._stdout = None
|
||||||
|
self._stderr = None
|
||||||
|
|
||||||
def _extract_text(self, result: Any) -> str:
|
def _extract_text(self, result: Any) -> str:
|
||||||
if isinstance(result, dict):
|
if isinstance(result, dict):
|
||||||
content = result.get("content")
|
content = result.get("content")
|
||||||
|
|||||||
@@ -336,16 +336,17 @@ class AIService:
|
|||||||
async def _fetch_mcp_tool_schemas(self) -> list[dict]:
|
async def _fetch_mcp_tool_schemas(self) -> list[dict]:
|
||||||
if self._mcp_tools_fetched:
|
if self._mcp_tools_fetched:
|
||||||
return self._mcp_tool_schemas or []
|
return self._mcp_tool_schemas or []
|
||||||
self._mcp_tools_fetched = True
|
|
||||||
if self.config.image_understand is None:
|
if self.config.image_understand is None:
|
||||||
log.info("MCP tools: image_understand not configured, skipping")
|
log.info("MCP tools: image_understand not configured, skipping")
|
||||||
self._mcp_tool_schemas = []
|
self._mcp_tool_schemas = []
|
||||||
|
self._mcp_tools_fetched = True
|
||||||
return []
|
return []
|
||||||
log.info("MCP tools: fetching tool list from %s", self.config.image_understand.command)
|
log.info("MCP tools: fetching tool list from %s", self.config.image_understand.command)
|
||||||
try:
|
try:
|
||||||
async with MinimaxImageUnderstandClient(self.config.image_understand) as mcp_client:
|
async with MinimaxImageUnderstandClient(self.config.image_understand) as mcp_client:
|
||||||
raw = await mcp_client.list_tools()
|
raw = await mcp_client.list_tools()
|
||||||
self._mcp_tool_schemas = mcp_tools_to_openai_schemas(raw)
|
self._mcp_tool_schemas = mcp_tools_to_openai_schemas(raw)
|
||||||
|
self._mcp_tools_fetched = True
|
||||||
names = [s["function"]["name"] for s in self._mcp_tool_schemas]
|
names = [s["function"]["name"] for s in self._mcp_tool_schemas]
|
||||||
log.info("MCP tools: loaded %d tool(s): %s", len(names), ", ".join(names) or "(none)")
|
log.info("MCP tools: loaded %d tool(s): %s", len(names), ", ".join(names) or "(none)")
|
||||||
except Exception:
|
except Exception:
|
||||||
|
|||||||
Reference in New Issue
Block a user