Mnemosyne MCP

Instructions

Implementation Reference

• src/neem/mcp/tools/graph_ops.py:189-218 (handler)

  The main handler function for the 'sparql_update' tool. It authenticates the user, validates the SPARQL query, submits an 'apply_update' job to the backend, waits for the job result using streaming or polling, and returns a JSON response with the job outcome.

  async def sparql_update_tool(
      sparql: str,
      context: Context | None = None,
  ) -> str:
      """Execute a SPARQL INSERT/DELETE/UPDATE operation."""
      auth = MCPAuthContext.from_context(context)
      auth.require_auth()
  
      if not sparql or not sparql.strip():
          raise ValueError("sparql update is required and cannot be empty")
  
      metadata = await submit_job(
          base_url=backend_config.base_url,
          auth=auth,
          task_type="apply_update",
          payload={"sparql": sparql.strip()},
      )
  
      if context:
          await context.report_progress(10, 100)
  
      result = await _wait_for_job_result(
          job_stream, metadata, context, auth
      )
  
      return _render_json({
          "success": True,
          "job_id": metadata.job_id,
          **result,
      })

• src/neem/mcp/tools/graph_ops.py:181-188 (registration)

  The @server.tool decorator registers the 'sparql_update' tool on the FastMCP server instance, providing the tool name, title, description, and input schema (inferred from the following function signature: sparql: str). This occurs within the register_graph_ops_tools function.

  @server.tool(
      name="sparql_update",
      title="Run SPARQL Update",
      description=(
          "Executes a SPARQL INSERT, DELETE, or UPDATE operation to modify graph data. "
          "Use this for adding, modifying, or removing triples from graphs."
      ),
  )

• src/neem/mcp/tools/graph_ops.py:223-278 (helper)

  Supporting helper function used by sparql_update_tool (and other graph tools) to wait for job completion, handling both WebSocket streaming and fallback polling, extracting results or errors.

  async def _wait_for_job_result(
      job_stream: Optional[RealtimeJobClient],
      metadata: JobSubmitMetadata,
      context: Optional[Context],
      auth: MCPAuthContext,
  ) -> JsonDict:
      """Wait for job completion via WebSocket or polling, return result info including detail."""
      events = None
      if job_stream and metadata.links.websocket:
          events = await stream_job(job_stream, metadata, timeout=STREAM_TIMEOUT_SECONDS)
  
      if events:
          if context:
              await context.report_progress(80, 100)
          # Check for completion status in events and extract result
          for event in reversed(events):
              event_type = event.get("type", "")
              if event_type in ("job_completed", "completed", "succeeded"):
                  if context:
                      await context.report_progress(100, 100)
                  # Extract result from event payload
                  result: JsonDict = {"status": "succeeded", "events": len(events)}
                  payload = event.get("payload", {})
                  if isinstance(payload, dict):
                      detail = payload.get("detail")
                      if detail:
                          result["detail"] = detail
                  return result
              if event_type in ("failed", "error"):
                  error = event.get("error", "Job failed")
                  return {"status": "failed", "error": error}
          return {"status": "unknown", "event_count": len(events)}
  
      # Fall back to polling
      status_payload = (
          await poll_job_until_terminal(metadata.links.status, auth)
          if metadata.links.status
          else None
      )
  
      if context:
          await context.report_progress(100, 100)
  
      if status_payload:
          status = status_payload.get("status", "unknown")
          detail = status_payload.get("detail")
          if status == "failed":
              error = status_payload.get("error") or (detail.get("error") if isinstance(detail, dict) else None)
              return {"status": "failed", "error": error}
          # Include full detail in result for successful jobs
          result: JsonDict = {"status": status}
          if detail:
              result["detail"] = detail
          return result
  
      return {"status": "unknown"}

• src/neem/mcp/server/standalone_server.py:316-316 (registration)

  Call to register_graph_ops_tools in the standalone MCP server creation, which includes registration of the sparql_update tool.

  register_graph_ops_tools(mcp_server)