rlm_sub_query

Instructions

Implementation Reference

• src/rlm_mcp_server.py:1549-1597 (handler)

  The actual handler function for the 'rlm_sub_query' tool. It makes a sub-LLM call on a chunk or filtered context. Resolves provider/model, loads context, optionally selects a chunk, then calls _make_provider_call and returns the response.

  @mcp.tool()
  async def rlm_sub_query(
      query: str,
      context_name: str,
      chunk_index: Optional[int] = None,
      provider: str = "auto",
      model: Optional[str] = None,
  ) -> dict:
      """Make a sub-LLM call on a chunk or filtered context. Core of recursive pattern.
  
      Args:
          query: Question/instruction for the sub-call
          context_name: Context identifier to query against
          chunk_index: Optional: specific chunk index
          provider: LLM provider - 'auto', 'ollama', or 'claude-sdk'. 'auto' prefers Ollama if available (free local inference)
          model: Model to use (provider-specific defaults apply)
      """
      # Resolve auto provider and model
      resolved_provider, resolved_model = await _resolve_provider_and_model(provider, model)
  
      error = _ensure_context_loaded(context_name)
      if error:
          return {"error": "context_not_found", "message": error}
  
      if chunk_index is not None:
          chunks = contexts[context_name].get("chunks")
          if not chunks or chunk_index >= len(chunks):
              return {"error": "chunk_not_available", "message": f"Chunk {chunk_index} not available"}
          context_content = chunks[chunk_index]
      else:
          context_content = contexts[context_name]["content"]
  
      result, call_error = await _make_provider_call(resolved_provider, resolved_model, query, context_content)
  
      if call_error:
          return {
              "error": "provider_error",
              "provider": resolved_provider,
              "model": resolved_model,
              "requested_provider": provider,
              "message": call_error,
          }
  
      return {
          "provider": resolved_provider,
          "model": resolved_model,
          "requested_provider": provider if provider == "auto" else None,
          "response": result,
      }

• src/rlm_mcp_server.py:1549-1549 (registration)

  Registration of 'rlm_sub_query' as a FastMCP tool via the @mcp.tool() decorator. This is the point where the tool is registered with the MCP server.

  @mcp.tool()

• src/rlm_mcp_server.py:1052-1067 (helper)

  Helper that resolves the 'auto' provider to either 'ollama' or 'claude-sdk', and selects the best model for the chosen provider.

  async def _resolve_provider_and_model(
      provider: str,
      model: Optional[str],
  ) -> tuple[str, str]:
      """Resolve 'auto' provider and get appropriate model."""
      # Handle auto provider selection
      if provider == "auto":
          # Check Ollama status (uses cache)
          await _check_ollama_status()
          provider = _get_best_provider()
  
      # Get model if not specified
      if not model:
          model = _get_best_model_for_provider(provider)
  
      return provider, model

• src/rlm_mcp_server.py:1070-1085 (helper)

  Helper that routes the sub-call to the appropriate provider (Ollama or Claude SDK) after resolving provider/model.

  async def _make_provider_call(
      provider: str,
      model: str,
      query: str,
      context_content: str,
  ) -> tuple[Optional[str], Optional[str]]:
      """Route a sub-call to the appropriate provider. Returns (result, error)."""
      # Resolve auto provider
      resolved_provider, resolved_model = await _resolve_provider_and_model(provider, model)
  
      if resolved_provider == "ollama":
          return await _call_ollama(query, context_content, resolved_model)
      elif resolved_provider == "claude-sdk":
          return await _call_claude_sdk(query, context_content, resolved_model)
      else:
          return None, f"Unknown provider: {resolved_provider}"

• src/rlm_mcp_server.py:976-987 (helper)

  Helper that ensures a context is loaded into memory (from disk if needed), returning an error string if the context doesn't exist.

  def _ensure_context_loaded(name: str) -> Optional[str]:
      """Ensure context is loaded into memory. Returns error message if not found."""
      if name in contexts:
          return None
  
      disk_context = _load_context_from_disk(name)
      if disk_context:
          content = disk_context.pop("content")
          contexts[name] = {"meta": disk_context, "content": content}
          return None
  
      return f"Context '{name}' not found"