Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 41 additions & 2 deletions blackduck/mcp_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -363,22 +363,61 @@ def list_project_components(project_name: str, version_name: Optional[str] = Non
licenses = component.get('licenses', [])
license_display = licenses[0].get('licenseDisplay', 'Unknown') if licenses else 'Unknown'

component_id = safe_get(component, 'component')
component_version_id = safe_get(component, 'componentVersion')

result.append({
'componentName': component.get('componentName'),
'componentVersionName': component.get('componentVersionName'),
'componentId': component_id.rstrip('/').split('/')[-1] if component_id else None,
'componentVersionId': component_version_id.rstrip('/').split('/')[-1] if component_version_id else None,
'matchTypes': component.get('matchTypes', []),
'usages': component.get('usages', []),
'licenseDisplay': license_display,
'policyStatus': component.get('policyStatus'),
'securityRiskProfile': component.get('securityRiskProfile'),
'activityData': component.get('activityData')
})

return result

except Exception as e:
logger.error(f"Error listing components: {e}")
raise

@self.mcp.tool
def get_upgrade_guidance(component_id: str, component_version_id: str) -> Dict[str, Any]:
"""Get BlackDuck's recommended safe upgrade versions for a component version.

Tries the transitive-aware upgrade-guidance endpoint first (accounts for
upgrade paths that also work for components pulled in transitively),
falling back to the standard upgrade-guidance endpoint if unavailable.

Args:
component_id: componentId from list_project_components
component_version_id: componentVersionId from list_project_components

Returns:
Dict with shortTerm/longTerm recommended upgrade versions, or an
'available': False marker if BlackDuck has no guidance for this version.
"""
base_path = f"/api/components/{component_id}/versions/{component_version_id}"
for suffix in ('/transitive-upgrade-guidance', '/upgrade-guidance'):
try:
guidance = self.client.get_json(base_path + suffix)
return {
'available': True,
'source': suffix.strip('/'),
'componentName': guidance.get('componentName'),
'versionName': guidance.get('versionName'),
'shortTerm': safe_get(guidance, 'shortTerm'),
'longTerm': safe_get(guidance, 'longTerm'),
}
except Exception as e:
logger.info(f"{suffix} unavailable for {component_id}/{component_version_id}: {e}")
continue

return {'available': False, 'componentId': component_id, 'componentVersionId': component_version_id}

def run(self):
"""Run the MCP server"""
Expand Down
91 changes: 91 additions & 0 deletions test/test_mcp_server_upgrade_guidance.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
import asyncio
from unittest.mock import patch

import pytest

# fastmcp is an optional 'mcp' extra, not part of the base test lockfile.
# Skip this module entirely rather than failing CI when it isn't installed.
pytest.importorskip("fastmcp")

from blackduck.mcp_server import BlackDuckMCPServer


@pytest.fixture
def mcp_server(monkeypatch):
monkeypatch.setenv("BLACKDUCK_URL", "https://fake-hub-host")
monkeypatch.setenv("BLACKDUCK_TOKEN", "fake-token")
with patch("blackduck.mcp_server.Client") as mock_client_cls:
server = BlackDuckMCPServer()
yield server, mock_client_cls.return_value


def _call_tool(server, name, **kwargs):
async def _run():
tool = await server.mcp.get_tool(name)
return tool.fn(**kwargs)
return asyncio.run(_run())


def test_get_upgrade_guidance_falls_back_when_transitive_variant_unavailable(mcp_server):
"""transitive-upgrade-guidance requires an originId that isn't always
known; the tool should fall back to the standard upgrade-guidance
endpoint rather than surfacing that as a failure."""
server, mock_client = mcp_server
standard_guidance = {
"componentName": "Apache Commons BeanUtils",
"versionName": "1.9.3",
"shortTerm": {"versionName": "1.11.0"},
"longTerm": {"versionName": "1.11.0"},
}
mock_client.get_json.side_effect = [Exception("404 not found"), standard_guidance]

result = _call_tool(
server, "get_upgrade_guidance",
component_id="comp-1", component_version_id="ver-1",
)

assert result["available"] is True
assert result["source"] == "upgrade-guidance"
assert result["shortTerm"] == {"versionName": "1.11.0"}
assert mock_client.get_json.call_count == 2
assert "transitive-upgrade-guidance" in mock_client.get_json.call_args_list[0].args[0]
assert mock_client.get_json.call_args_list[1].args[0].endswith("/upgrade-guidance")


def test_get_upgrade_guidance_reports_unavailable_when_both_endpoints_fail(mcp_server):
server, mock_client = mcp_server
mock_client.get_json.side_effect = [Exception("404"), Exception("404")]

result = _call_tool(
server, "get_upgrade_guidance",
component_id="comp-1", component_version_id="ver-1",
)

assert result == {"available": False, "componentId": "comp-1", "componentVersionId": "ver-1"}


def test_list_project_components_extracts_ids_from_hrefs(mcp_server):
server, mock_client = mcp_server

def get_resource(name, parent=None, **kwargs):
if name == "projects":
return iter([{"name": "Acme", "_meta": {"href": "https://fake-hub-host/api/projects/p1"}}])
if name == "versions":
return iter([{"versionName": "1.0", "_meta": {"href": "https://fake-hub-host/api/projects/p1/versions/v1"}}])
if name == "components":
return iter([{
"componentName": "ag-grid-community",
"componentVersionName": "24.0.0",
"component": "https://fake-hub-host/api/components/comp-1",
"componentVersion": "https://fake-hub-host/api/components/comp-1/versions/ver-1",
"matchTypes": ["DIRECT_DEPENDENCY"],
}])
raise AssertionError(f"unexpected resource name: {name}")

mock_client.get_resource.side_effect = get_resource

result = _call_tool(server, "list_project_components", project_name="Acme")

assert result[0]["componentId"] == "comp-1"
assert result[0]["componentVersionId"] == "ver-1"
assert result[0]["matchTypes"] == ["DIRECT_DEPENDENCY"]