19. Query a database from a tool

search_applicants should hit the real database, filter by what the LLM asked for, and return only the columns this user is allowed to see.

Problem. search_applicants should hit the real database, filter by what the LLM asked for, and return only the columns this user is allowed to see.

Solution. You're inside Rails — use ActiveRecord. Build the query with the relation API (which parameterizes for you), then map the records onto your @rbs type shape explicitly. Gate sensitive columns with @requires on an output variant.

# app/service/workflows/search_applicants.rb
module Workflows
  class SearchApplicants
    # @rbs import error

    include McpAuthorization::DSL

    # @rbs type applicant_summary = {
    #   id: String,
    #   name: String,
    #   stage: String
    # }

    # @rbs type pii_summary = {
    #   id: String,
    #   name: String,
    #   stage: String,
    #   email: String,
    #   phone: String
    # }

    # @rbs type success     = { success: true, total: Integer, applicants: Array[applicant_summary] }
    # @rbs type pii_success = { success: true, total: Integer, applicants: Array[pii_summary] }
    # Gate the whole VARIANT, not a field inside it — and list it first (see Result).
    # @rbs type output = pii_success @requires(:view_pii)
    #                  | success
    #                  | error

    def description
      "Search applicants by name fragment and/or current stage."
    end

    #: (
    #:   ?query: String?   @min(1) @max(100) @desc(Case-insensitive name fragment),
    #:   ?stage: String?   @desc(Exact stage to filter by),
    #:   ?limit: Integer   @min(1) @max(100) @default(20)
    #: ) -> Hash[Symbol, untyped]
    def call(query: nil, stage: nil, limit: 20)
      scope = Applicant.all
      # Relation methods parameterize the value — the LLM's text never touches raw SQL.
      scope = scope.where("name ILIKE ?", "%#{query}%") if query.present?  # ILIKE is PostgreSQL-only; use LIKE on MySQL/SQLite (case-sensitivity varies by collation)
      scope = scope.where(stage: stage) if stage.present?

      total   = scope.count
      records = scope.order(updated_at: :desc).limit(limit).to_a

      {
        success: true,
        total: total,
        applicants: records.map { |a| summarize(a) }
      }
    rescue ActiveRecord::StatementInvalid => e
      { success: false, error: { code: "query_failed", message: e.message,
                                 hint: "Check the stage value against list_workflow_stages." } }
    end

    private

    def summarize(applicant)
      # id is declared `String` in the type — to_s it, since projection
      # passes values through without coercing them.
      base = { id: applicant.id.to_s, name: applicant.name, stage: applicant.stage }
      return base unless can?(:view_pii)

      base.merge(email: applicant.email, phone: applicant.phone)
    end
  end
end

Result. The query is parameterized, so a query: of "'; DROP TABLE applicants; --" is matched as a literal name fragment, not executed. PII is protected in two layers. The source of truth is summarize: it only adds email/phone when can?(:view_pii). The schema is the backstop — but only if you gate the right thing. pii_success is a @requires(:view_pii) variant (the tag sits on the union member, not on a field inside the named type — a field-level tag on a variant resolved by name is honored for nesting but the wrong tool for "show this whole shape to some users"). For a user without the flag, the variant is dropped from the output schema entirely and filter_output projects the return value onto the remaining success shape, stripping email/phone before serialization even if a handler bug let them through. The variant is listed first because success and pii_success share the same top-level keys: when both are visible (a view_pii user), the gem breaks the tie by source order, so the richer PII shape must come first or it gets projected away. Pagination is bounded by @max(100) so the LLM can't ask for a million rows.

Variant: a second database or raw SQL. Reading from an analytics replica or a non-AR datastore? Borrow a pooled connection and sanitize explicitly — never string-build with LLM input:

def call(account_id:, since:)
  rows = ApplicationRecord.connection.exec_query(
    ApplicationRecord.sanitize_sql_array(
      ["SELECT stage, COUNT(*) AS n FROM events WHERE account_id = ? AND created_at >= ? GROUP BY stage",
       account_id, since]
    )
  )
  { success: true, by_stage: rows.map { |r| { stage: r["stage"], count: r["n"] } } }
end

The connection comes from ActiveRecord's pool and is returned automatically at the end of the request — you don't open or close it yourself.

Collected from COOKBOOK.md in the repository. Edit it there, not here.