mcp_authorization

Rails engine for serving MCP tools with per-request schema discrimination compiled from RBS type annotations.

Add it to your Gemfile and your Rails app speaks MCP. Write @rbs type comments in plain Ruby service classes, tag fields and variants with @requires(:flag), and the gem compiles tailored JSON Schema per request. The type definitions are the authorization policy.

The gem gives you three independent controls over what each user sees:

All three go through the same predicate: current_user.can?(:symbol). The symbol can represent a permission, a feature flag, a plan tier, an A/B bucket -- whatever your app puts behind it.

# Gemfile
gem "mcp_authorization"

bundle install

Routes install automatically at /mcp. No mount needed.

# config/initializers/mcp_authorization.rb
McpAuthorization.configure do |config|
  config.server_name    = "my-app"
  config.server_version = "1.0.0"

  # Build a context from each MCP request.
  # Return anything that responds to .current_user.can?(symbol).
  config.context_builder = ->(request) {
    user = User.authenticate(request.headers["Authorization"])
    OpenStruct.new(current_user: user)
  }
end

The gem has two opinions about your app:

context.current_user.can?(:symbol)         # => true/false  (required)
context.current_user.default_for(:symbol)  # => value | nil (optional)

can? gates visibility -- fields, variants, and entire tools. default_for populates JSON Schema default values from the current user's context. The symbols can mean anything:

current_user.can?(:manage_workflows)  # permission
current_user.can?(:backward_routing)  # feature flag
current_user.can?(:enterprise_plan)   # plan tier
current_user.can?(:experiment_v2)     # A/B test

current_user.default_for(:timezone)   # => "America/Chicago"
current_user.default_for(:locale)     # => "en-US"

default_for is optional. If you don't use @default_for tags, you don't need it. When present, it's a simple case statement -- no metaprogramming:

def default_for(key)
  case key
  when :timezone then timezone
  when :locale then locale
  end
end

Define reusable types as .rbs files. These are plain RBS -- no comment markers.

# sig/shared/error.rbs
type error_code = "not_found"
               | "invalid_transition"
               | "already_at_stage"

type error = {
  success: false,
  error: { code: error_code, message: String, hint: String }
}

# sig/shared/applicant.rbs
type applicant = {
  id: String,
  name: String,
  current_stage: String,
  applied_at: String
}

A handler includes McpAuthorization::DSL, imports shared types, and defines its own types. The #: annotation on def call is the input schema -- tag params with @requires to control who sees them.

# app/service/workflows/advance_step.rb
module Workflows
  class AdvanceStep
    # @rbs import error

    include McpAuthorization::DSL

    # @rbs type success = {
    #   success: true,
    #   applicant_id: String,
    #   current_stage: String
    # }

    # @rbs type rerouted_success = {
    #   success: true,
    #   applicant_id: String,
    #   previous_stage: String,
    #   current_stage: String,
    #   audit_trail: Array[String]
    # }

    # @rbs type output = success
    #                   | rerouted_success  @requires(:backward_routing)
    #                   | error

    def description
      if can?(:backward_routing)
        "Advance an applicant to any stage, or reroute them backward."
      else
        "Advance an applicant to the next stage."
      end
    end

    #: (
    #:   applicant_id: String,
    #:   workflow_id: String,
    #:   ?stage_id: String?    @requires(:backward_routing),
    #:   ?reason: String?      @requires(:backward_routing)
    #: ) -> Hash[Symbol, untyped]
    def call(applicant_id:, workflow_id:, stage_id: nil, reason: nil)
      # your logic here
    end
  end
end

# app/mcp/workflows/advance_step_tool.rb
module Workflows
  class AdvanceStepTool < McpAuthorization::Tool
    tool_name "advance_step"
    authorization :manage_workflows
    not_destructive!
    tags "operator"
    dynamic_contract Workflows::AdvanceStep
  end
end

A user without :backward_routing:

advance_step — "Advance an applicant to the next stage."
  input:  applicant_id, workflow_id
  output: success | error

A user with :backward_routing:

advance_step — "Advance an applicant to any stage, or reroute them backward."
  input:  applicant_id, workflow_id, stage_id, reason
  output: success | rerouted_success | error

Same tool, same endpoint. The feature flag shapes the schema.

A handler includes McpAuthorization::DSL and implements two methods:

The DSL mixin provides initialize(server_context:), server_context, and can?(:flag).

The input schema is inferred from the #: annotation on def call. The output schema comes from @rbs type output. No separate schema definition needed.

On input params -- the param is excluded from the input schema when can? returns false. Tag them in the #: annotation above def call:

#: (
#:   query: String,
#:   ?force: bool            @requires(:admin),
#:   ?include_deleted: bool  @requires(:admin)
#: ) -> Hash[Symbol, untyped]
def call(query:, force: false, include_deleted: false)

On output variants -- the variant is excluded from the oneOf:

# @rbs type output = public_result
#                   | admin_result  @requires(:admin)
#                   | error

Untagged params and variants are always included.

Define reusable types as .rbs files in sig/shared/ (configurable via shared_type_paths):

# sig/shared/pagination.rbs
type pagination = {
  page: Integer,
  per_page: Integer,
  total: Integer
}

Import them in any handler:

# @rbs import pagination
# @rbs import error

# @rbs type success = {
#   success: true,
#   items: Array[String],
#   pagination: pagination
# }

# @rbs type output = success | error

The compiler loads sig/shared/pagination.rbs and sig/shared/error.rbs, parses their type definitions, and merges them into the handler's type map. The handler's own @rbs type definitions override on conflict.

Shared types define shapes. Authorization (@requires) stays on the handler -- it's a local policy decision, not a property of the type itself.

class MyTool < McpAuthorization::Tool
  tool_name "my_tool"
  authorization :some_flag        # tool hidden when can?(:some_flag) is false
  tags "recruiting", "operations" # which domains this tool appears in
  read_only!                      # MCP annotation hints
  dynamic_contract MyService      # handler class
end

Tools self-register when loaded. Put them anywhere under tool_paths (default: app/mcp/).

If a handler is missing required methods or schema definitions, the gem raises an ArgumentError on first request with a full diagnostic:

MyHandler does not satisfy the McpAuthorization handler contract.

Problems:
  - missing instance method #call
  - missing instance method #description
  - missing output schema (define # @rbs type output = variant1 | variant2 | ...)

A handler class should look like:

  class MyHandler
    include McpAuthorization::DSL

    # @rbs type output = success | error

    def description
      "What this tool does"
    end

    #: (name: String, ?force: bool @requires(:admin)) -> Hash[Symbol, untyped]
    def call(name:, force: false)
      # ...
    end
  end

POST /mcp/operator    -> tools tagged "operator"
POST /mcp/recruiting  -> tools tagged "recruiting"
POST /mcp             -> tools tagged with default_domain

Tag a tool with multiple domains to make it available in each:

tags "operator", "recruiting"

The @rbs type comments compile to JSON Schema:

# Primitives
# @rbs type x = String    -> { "type": "string" }
# @rbs type x = Integer   -> { "type": "integer" }
# @rbs type x = Float     -> { "type": "number" }
# @rbs type x = bool      -> { "type": "boolean" }
# @rbs type x = true      -> { "type": "boolean", "const": true }
# @rbs type x = false     -> { "type": "boolean", "const": false }

# String enums
# @rbs type status = "pending"
#                  | "active"
#                  | "closed"

# Records
# @rbs type result = {
#   success: bool,
#   message: String,
#   count?: Integer
# }
# (count? is optional -- excluded from "required")

# Arrays
# @rbs type items = Array[String]

# Type references (resolved from local types and imports)
# @rbs type input = { id: String, status: status }

Tag any field in a #: annotation or @rbs type record to add JSON Schema constraints. Tags are written as @tag(value) after the type:

#: (
#:   name: String                          @min(1) @max(100),
#:   email: String                         @format(email),
#:   age: Integer                          @min(0) @max(150),
#:   score: Float                          @exclusive_min(0) @exclusive_max(1.0),
#:   tags: Array[String]                   @min(1) @max(10) @unique(),
#:   quantity: Integer                     @multiple_of(5),
#:   ?timezone: String                     @default_for(:timezone),
#:   ?stage_id: String?                    @requires(:backward_routing) @depends_on(:workflow_id)
#: ) -> Hash[Symbol, untyped]

Value constraints:

Metadata:

Authorization:

Niche:

The @min / @max tags are type-aware: on strings they emit minLength/maxLength, on numbers minimum/maximum, and on arrays minItems/maxItems.

The #: annotation above def call supports multiple lines. Each line starts with #::

#: (
#:   applicant_id: String       @desc(Use fetch_latest_applicant to find this),
#:   workflow_id: String,
#:   ?stage_id: String?         @requires(:backward_routing) @depends_on(:workflow_id),
#:   ?reason: String?           @requires(:backward_routing)
#: ) -> Hash[Symbol, untyped]
def call(applicant_id:, workflow_id:, stage_id: nil, reason: nil)

Prefix a param with ? to mark it optional. Suffix the type with ? for nilable types. Both together (?name: Type?) means the field is optional and can be nil.

Use @depends_on(:parent_field) to express that a field is only required when another field is present. This emits JSON Schema dependentRequired:

#: (
#:   workflow_id: String,
#:   ?stage_id: String?      @requires(:backward_routing) @depends_on(:workflow_id),
#:   ?reason: String?        @requires(:backward_routing) @depends_on(:stage_id)
#: ) -> Hash[Symbol, untyped]

When :backward_routing is enabled, the schema includes:

{
  "dependentRequired": {
    "workflow_id": ["stage_id"],
    "stage_id": ["reason"]
  }
}

Literal true / false types become "const" values in JSON Schema:

# @rbs type success = { success: true, data: String }
# @rbs type error   = { success: false, code: String }
# @rbs type output  = success | error

MCP clients can narrow on success: const true vs success: const false -- the same pattern as TypeScript discriminated unions.

Source files are parsed once at boot and cached in memory. Only @requires filtering runs per request (hash lookups and can? calls). In development, caches are cleared automatically on file change via the Rails reloader.

In development mode, the gem wires into the Rails reloader. Edit an @rbs type annotation, save, and the next MCP request returns the updated schema. No server restart needed.

# List tools visible to a given role
bundle exec rake "mcp:tools[operator,manager]"

# Print Claude Code / Claude Desktop config JSON
bundle exec rake "mcp:claude[operator,manager]"

# Launch MCP Inspector (requires npx)
bundle exec rake "mcp:inspect[operator,manager]"

Rake tasks require cli_context_builder:

config.cli_context_builder = ->(domain:, role:) {
  user = User.new(role: role, permissions: ROLE_PERMISSIONS[role])
  OpenStruct.new(current_user: user)
}

1. MCP client sends a request to /mcp/:domain
2. Engine calls your context_builder with the request
3. ToolRegistry filters tools by domain tag and authorization gate (can? check)
4. RbsSchemaCompiler loads shared types from # @rbs import declarations
5. Input schema is compiled from the #: annotation on def call, filtering @requires params
6. Output schema is compiled from @rbs type output, filtering @requires variants
7. MCP client receives tool definitions with schemas tailored to the current user

Different users hitting the same endpoint can see different tools, different descriptions, different input fields, and different output shapes.

The gem uses the MCP SDK's Streamable HTTP transport in stateless mode. Each HTTP request creates a fresh MCP::Server, materialized with tools filtered and shaped for the current user. There is no persistent session or SSE stream between requests.

This is a deliberate choice. The gem's value is per-request schema discrimination -- the same endpoint returns different JSON Schema depending on who's asking. A stateful session would bake the tool list at connection time, meaning permission changes during a session would serve stale schemas until reconnect.

In practice this doesn't matter because MCP clients call tools/list once -- at the start of a conversation or when manually refreshed. The schema returned at that point is what the client (and the LLM behind it) uses for the entire conversation. Tool calls made later in the conversation still go through context_builder and the authorization gate, so a revoked permission results in a rejected call, not a leaked capability.

The tradeoff: stateless mode cannot send notifications/tools/list_changed or use report_progress during long-running tool calls, since both require an open SSE stream. For most use cases this is the right default -- schemas that reflect the current user's permissions at conversation start, enforced again at call time.

• Ruby >= 3.1
• Rails >= 6.0
• mcp ~> 0.10

MIT