sparql_query
Execute SPARQL SELECT or CONSTRUCT queries to search and retrieve data from knowledge graphs, returning results as JSON for data analysis and integration.
Instructions
Executes a read-only SPARQL SELECT or CONSTRUCT query against the user's graphs. Returns query results as JSON. Use this for searching and retrieving data from graphs.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| sparql | Yes | ||
| result_format | No | json |
Implementation Reference
- src/neem/mcp/tools/graph_ops.py:137-179 (handler)Main handler function for the sparql_query tool. Submits a 'run_query' job to the backend with the SPARQL query and result_format, waits for result, extracts query results if available, and returns formatted JSON.async def sparql_query_tool( sparql: str, result_format: str = "json", context: Context | None = None, ) -> str: """Execute a SPARQL SELECT/CONSTRUCT query.""" auth = MCPAuthContext.from_context(context) auth.require_auth() if not sparql or not sparql.strip(): raise ValueError("sparql query is required and cannot be empty") metadata = await submit_job( base_url=backend_config.base_url, auth=auth, task_type="run_query", payload={ "sparql": sparql.strip(), "result_format": result_format, }, ) if context: await context.report_progress(10, 100) result = await _wait_for_job_result( job_stream, metadata, context, auth ) # Extract query results from job output query_result = _extract_query_result(result) if query_result is not None: return _render_json({ "success": True, "results": query_result, "job_id": metadata.job_id, }) return _render_json({ "success": True, "job_id": metadata.job_id, **result, })
- src/neem/mcp/tools/graph_ops.py:129-136 (registration)Registers the sparql_query tool on the FastMCP server instance with name, title, and description.@server.tool( name="sparql_query", title="Run SPARQL Query", description=( "Executes a read-only SPARQL SELECT or CONSTRUCT query against the user's graphs. " "Returns query results as JSON. Use this for searching and retrieving data from graphs." ), )
- src/neem/mcp/server/standalone_server.py:316-316 (registration)Calls register_graph_ops_tools to register the graph operations tools including sparql_query on the MCP server.register_graph_ops_tools(mcp_server)
- Helper function to extract SPARQL query results from the job result dictionary, handling the specific backend response format.def _extract_query_result(result: JsonDict) -> Optional[Any]: """Extract SPARQL query results from job output. The backend returns query results as: - detail.result_inline.raw for SPARQL SELECT/CONSTRUCT queries - detail.result_inline for other operations """ if "detail" not in result or not isinstance(result["detail"], dict): return None detail = result["detail"] inline = detail.get("result_inline") if inline is None: return None # SPARQL query results are wrapped in {"raw": actual_result} if isinstance(inline, dict) and "raw" in inline: return inline["raw"] return inline