Skip to content
Merged
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
Original file line number Diff line number Diff line change
Expand Up @@ -570,6 +570,58 @@ async def run_sampler_tool(
)


##################################################
## MCP Prompts
## - https://modelcontextprotocol.io/docs/concepts/prompts
##################################################


@mcp.prompt()
def run_bell_state(backend_name: str = "") -> str:
"""Run a Bell state circuit on an IBM Quantum backend and interpret the results."""
backend_clause = (
f"on backend '{backend_name}'"
if backend_name
else "on the least busy backend (use least_busy_backend_tool to find it)"
)
return (
f"Run a Bell state circuit {backend_clause}: "
"1) Read the circuits://bell-state resource to get the circuit, "
f"2) Call run_sampler_tool with the circuit QPY and backend_name='{backend_name}', "
"3) Call get_job_status_tool with the returned job_id until status is DONE, "
"4) Call get_job_results_tool to retrieve measurement counts, "
"5) Interpret the results - expect approximately 50% '00' and 50% '11' outcomes."
)


@mcp.prompt()
def explore_backend(backend_name: str) -> str:
"""Explore an IBM Quantum backend's properties, calibration, and connectivity."""
return (
f"Explore the '{backend_name}' IBM Quantum backend: "
f"1) Call get_backend_properties_tool with backend_name='{backend_name}' "
"to get static properties (qubits, gates, processor type), "
f"2) Call get_backend_calibration_tool with backend_name='{backend_name}' "
"to get T1/T2 times and error rates, "
f"3) Call get_coupling_map_tool with backend_name='{backend_name}' "
"for qubit connectivity, "
"4) Summarize the backend's key characteristics and any notable calibration issues."
)


@mcp.prompt()
def monitor_job(job_id: str) -> str:
"""Monitor a running IBM Quantum job and retrieve its results when complete."""
return (
f"Monitor job '{job_id}' and retrieve results: "
f"1) Call get_job_status_tool with job_id='{job_id}', "
f"2) If status is DONE, call get_job_results_tool with job_id='{job_id}' "
"to get measurement counts, "
"3) If status is ERROR, report the error details from the status response, "
"4) If still running, report the current status and suggest checking again shortly."
)


# Resources
@mcp.resource("ibm://status", mime_type="text/plain")
async def get_service_status_resource() -> str:
Expand Down Expand Up @@ -630,6 +682,24 @@ def get_superposition_resource() -> dict[str, Any]:
return get_superposition_circuit()


##################################################
## MCP Resource Templates
## - https://modelcontextprotocol.io/docs/concepts/resources#resource-templates
##################################################


@mcp.resource("ibm://backends/{backend_name}", mime_type="application/json")
async def backend_properties_resource(backend_name: str) -> dict[str, Any]:
"""Get properties for a specific IBM Quantum backend."""
return await get_backend_properties(backend_name)


@mcp.resource("ibm://jobs/{job_id}", mime_type="application/json")
async def job_status_resource(job_id: str) -> dict[str, Any]:
"""Get the status of a specific IBM Quantum job."""
return await get_job_status(job_id)


def main() -> None:
"""Run the server."""
mcp.run(transport="stdio", show_banner=False)
Expand Down
54 changes: 54 additions & 0 deletions qiskit-ibm-runtime-mcp-server/tests/test_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
setup_ibm_quantum_account,
usage_info,
)
from qiskit_ibm_runtime_mcp_server.server import mcp
from qiskit_ibm_runtime_mcp_server.server import (
active_account_info_tool,
active_instance_info_tool,
Expand Down Expand Up @@ -2627,4 +2628,57 @@ def test_all_circuits_have_usage_instructions(self):
assert "run_sampler_tool" in circuit["usage"]


class TestServerRegistration:
"""Test that tools, resources, prompts, and templates are registered."""

def test_server_name(self):
"""Test the server name is correct."""
assert mcp.name == "Qiskit IBM Runtime"

def test_resources_registered(self):
"""Test that all expected static resources are registered."""
resource_uris = set(mcp._resource_manager._resources.keys())
expected_resources = {
"ibm://status",
"circuits://bell-state",
"circuits://ghz-state",
"circuits://random",
"circuits://superposition",
}
assert expected_resources.issubset(resource_uris), (
f"Missing resources: {expected_resources - resource_uris}"
)

def test_resource_count(self):
"""Test the expected number of static resources."""
assert len(mcp._resource_manager._resources) == 5

def test_prompts_registered(self):
"""Test that all expected prompts are registered."""
prompt_names = set(mcp._prompt_manager._prompts.keys())
expected_prompts = {"run_bell_state", "explore_backend", "monitor_job"}
assert expected_prompts.issubset(prompt_names), (
f"Missing prompts: {expected_prompts - prompt_names}"
)

def test_prompt_count(self):
"""Test the expected number of prompts."""
assert len(mcp._prompt_manager._prompts) == 3

def test_resource_templates_registered(self):
"""Test that all expected resource templates are registered."""
template_uris = set(mcp._resource_manager._templates.keys())
expected_templates = {
"ibm://backends/{backend_name}",
"ibm://jobs/{job_id}",
}
assert expected_templates.issubset(template_uris), (
f"Missing resource templates: {expected_templates - template_uris}"
)

def test_resource_template_count(self):
"""Test the expected number of resource templates."""
assert len(mcp._resource_manager._templates) == 2


# Assisted by watsonx Code Assistant
Loading