feat: initial v0.1.0 of mcp-sonarqube-proxy

Stdio MCP server that proxies tools from an upstream SonarQube MCP server
over streamable HTTP. Tools are forwarded 1:1 with full schema preservation
(inputSchema, outputSchema, annotations, title, _meta); CallToolResult is
forwarded including isError and structuredContent.

- proxy.py: persistent upstream ClientSession, low-level Server with
  @list_tools and @call_tool(validate_input=False) handlers — the upstream
  is the sole schema authority.
- cli.py: Click-based 'serve' (stdio) and 'check' (probe) commands;
  logging strictly on stderr (stdout reserved for JSON-RPC).
- Targets mcp 1.27.x decorator API (pinned <2 to guard against the
  unreleased constructor-API rewrite on main).
- pytest suite (14 tests) covering env-var resolution, schema passthrough,
  CallToolResult forwarding, registration, dispatch end-to-end, and CLI
  success/error paths.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-06 20:26:38 +02:00
commit a6fd188c14
10 changed files with 647 additions and 0 deletions
+150
View File
@@ -0,0 +1,150 @@
"""Tests for the proxy core: discovery, schema preservation, call forwarding."""
from __future__ import annotations
from typing import Any
from unittest.mock import AsyncMock
import pytest
from mcp import types
from mcp_sonarqube_proxy.proxy import (
DEFAULT_UPSTREAM_URL,
build_proxy_server,
call_tool_via,
list_tools_via,
upstream_url,
)
def _full_tool() -> types.Tool:
"""A Tool with every passthrough field populated."""
return types.Tool(
name="rich_tool",
title="Rich Tool",
description="A tool with full metadata.",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "search query"},
"limit": {"type": "integer", "minimum": 1, "default": 10},
},
"required": ["query"],
"additionalProperties": False,
},
outputSchema={
"type": "object",
"properties": {"hits": {"type": "array", "items": {"type": "string"}}},
},
annotations=types.ToolAnnotations(
title="Rich Tool",
readOnlyHint=True,
idempotentHint=True,
openWorldHint=False,
),
)
def _fake_upstream(*, list_tools_return: Any = None, call_tool_return: Any = None) -> AsyncMock:
upstream = AsyncMock()
if list_tools_return is not None:
upstream.list_tools.return_value = list_tools_return
if call_tool_return is not None:
upstream.call_tool.return_value = call_tool_return
return upstream
def test_upstream_url_default(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("SONARQUBE_MCP_URL", raising=False)
assert upstream_url() == DEFAULT_UPSTREAM_URL
def test_upstream_url_env_override(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("SONARQUBE_MCP_URL", "http://example.test:9000/mcp")
assert upstream_url() == "http://example.test:9000/mcp"
async def test_list_tools_via_returns_full_tool_objects() -> None:
tool = _full_tool()
upstream = _fake_upstream(list_tools_return=types.ListToolsResult(tools=[tool]))
tools = await list_tools_via(upstream)
assert len(tools) == 1
# Round-trip dump must match exactly: schemas, annotations, title all survive.
assert tools[0].model_dump(by_alias=True) == tool.model_dump(by_alias=True)
upstream.list_tools.assert_awaited_once()
async def test_list_tools_via_returns_empty_list() -> None:
upstream = _fake_upstream(list_tools_return=types.ListToolsResult(tools=[]))
tools = await list_tools_via(upstream)
assert tools == []
async def test_list_tools_via_propagates_errors() -> None:
upstream = AsyncMock()
upstream.list_tools.side_effect = RuntimeError("upstream gone")
with pytest.raises(RuntimeError, match="upstream gone"):
await list_tools_via(upstream)
async def test_call_tool_via_forwards_full_result() -> None:
expected = types.CallToolResult(
content=[types.TextContent(type="text", text="ok")],
structuredContent={"score": 42},
isError=False,
)
upstream = _fake_upstream(call_tool_return=expected)
result = await call_tool_via(upstream, "do_thing", {"x": 1})
assert result is expected
upstream.call_tool.assert_awaited_once_with("do_thing", {"x": 1})
async def test_call_tool_via_preserves_is_error_results() -> None:
expected = types.CallToolResult(
content=[types.TextContent(type="text", text="boom")],
isError=True,
)
upstream = _fake_upstream(call_tool_return=expected)
result = await call_tool_via(upstream, "broken", {})
assert result.isError is True
assert result.content[0].text == "boom"
async def test_call_tool_via_propagates_errors() -> None:
upstream = AsyncMock()
upstream.call_tool.side_effect = ConnectionError("link down")
with pytest.raises(ConnectionError, match="link down"):
await call_tool_via(upstream, "anything", {})
def test_build_proxy_server_registers_handlers() -> None:
upstream = AsyncMock()
server = build_proxy_server(upstream)
assert types.ListToolsRequest in server.request_handlers
assert types.CallToolRequest in server.request_handlers
async def test_build_proxy_server_handler_forwards_tools_through_dispatch() -> None:
"""End-to-end: dispatch a ListToolsRequest through the registered handler."""
tool = _full_tool()
upstream = _fake_upstream(list_tools_return=types.ListToolsResult(tools=[tool]))
server = build_proxy_server(upstream)
handler = server.request_handlers[types.ListToolsRequest]
server_result = await handler(types.ListToolsRequest(method="tools/list"))
inner = server_result.root # ServerResult wraps a ListToolsResult
assert isinstance(inner, types.ListToolsResult)
assert len(inner.tools) == 1
assert inner.tools[0].model_dump(by_alias=True) == tool.model_dump(by_alias=True)