forked from HAL9000/cleveragents-core
freemo
b96138b88e
MCPToolAdapter.invoke() was reading error messages from result.get('error',
'unknown error'), but the MCP 1.4.0 protocol returns errors in the content
field as a list of content items with type and text keys. This caused every
error from a real MCP 1.4.0-compliant server to be silently replaced with
the string 'unknown error'.
Changes:
- src/cleveragents/mcp/adapter.py: extract error_text from content[0].text
with safe guards (isinstance check, length check) and fallback to
'unknown error' when content is absent or empty
- features/mocks/mock_mcp_transport.py: return MCP 1.4.0-compliant error
responses using content list format instead of the non-standard error key
- features/tdd_mcp_error_content_key.feature: Behave scenario verifying
correct error extraction from MCP 1.4.0 content arrays (written as TDD
issue-capture, @tdd_expected_fail removed after fix applied)
- features/steps/tdd_mcp_error_content_key_steps.py: step definitions for
the new scenario including _MCP14ErrorTransport mock subclass
All 51 MCP adapter scenarios pass. Typecheck: 0 errors. Lint: clean.
ISSUES CLOSED: #2158
71 lines
2.2 KiB
Python
71 lines
2.2 KiB
Python
"""Shared mock MCP transport for Behave and Robot Framework tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from cleveragents.mcp.adapter import MCPServerConfig, MCPTransport
|
|
|
|
|
|
class MockMCPTransport(MCPTransport):
|
|
"""In-memory mock transport simulating an MCP server.
|
|
|
|
Supports configurable tools, connection failures, invocation results,
|
|
invocation errors, and tool-level timeouts for deterministic testing.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
tools: list[dict[str, Any]] | None = None,
|
|
*,
|
|
fail_connect: bool = False,
|
|
invoke_results: dict[str, dict[str, Any]] | None = None,
|
|
invoke_errors: dict[str, str] | None = None,
|
|
timeout_tools: set[str] | None = None,
|
|
) -> None:
|
|
self._tools = tools or []
|
|
self._fail_connect = fail_connect
|
|
self._invoke_results = invoke_results or {}
|
|
self._invoke_errors = invoke_errors or {}
|
|
self._timeout_tools = timeout_tools or set()
|
|
self._connected = False
|
|
|
|
def connect(self, config: MCPServerConfig) -> dict[str, Any]:
|
|
if self._fail_connect:
|
|
msg = "Mock connection refused"
|
|
raise ConnectionRefusedError(msg)
|
|
self._connected = True
|
|
return {"capabilities": {"tools": True}}
|
|
|
|
def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
|
|
if method == "tools/list":
|
|
return {"tools": list(self._tools)}
|
|
|
|
if method == "tools/call":
|
|
tool_name = params.get("name", "")
|
|
|
|
if tool_name in self._timeout_tools:
|
|
msg = f"Tool '{tool_name}' exceeded timeout"
|
|
raise TimeoutError(msg)
|
|
|
|
if tool_name in self._invoke_errors:
|
|
return {
|
|
"isError": True,
|
|
"content": [
|
|
{"type": "text", "text": self._invoke_errors[tool_name]}
|
|
],
|
|
}
|
|
|
|
if tool_name in self._invoke_results:
|
|
return {"content": self._invoke_results[tool_name]}
|
|
|
|
return {"content": {"result": "ok"}}
|
|
|
|
return {}
|
|
|
|
def close(self) -> None:
|
|
self._connected = False
|
|
|
|
def add_tool(self, tool: dict[str, Any]) -> None:
|
|
self._tools.append(tool)
|