Module: Tina4::McpDevTools

Defined in:

lib/tina4/mcp.rb

Overview

── Built-in dev tools ────────────────────────────────────────────

Class Method Summary

• .register(server) ⇒ Object

  Register all 24 built-in dev tools on the given McpServer.

Class Method Details

.register(server) ⇒ Object

Register all 24 built-in dev tools on the given McpServer.

# File 'lib/tina4/mcp.rb', line 432

def self.register(server)
  project_root = File.expand_path(Dir.pwd)

  # ── Helpers ────────────────────────────────────────
  safe_path = lambda do |rel_path|
    resolved = File.expand_path(rel_path, project_root)
    unless resolved.start_with?(project_root)
      raise ArgumentError, "Path escapes project directory: #{rel_path}"
    end
    resolved
  end

  redact_env = lambda do |key, value|
    sensitive = %w[secret password token key credential api_key]
    if sensitive.any? { |s| key.downcase.include?(s) }
      "***REDACTED***"
    else
      value
    end
  end

  # ── Database Tools ────────────────────────────────
  server.register_tool("database_query", lambda { |sql:, params: "[]"|
    db = Tina4.database
    return { "error" => "No database connection" } if db.nil?
    param_list = params.is_a?(String) ? JSON.parse(params) : params
    result = db.fetch(sql, param_list)
    { "records" => result.to_a, "count" => result.count }
  }, "Execute a read-only SQL query (SELECT)")

  server.register_tool("database_execute", lambda { |sql:, params: "[]"|
    db = Tina4.database
    return { "error" => "No database connection" } if db.nil?
    param_list = params.is_a?(String) ? JSON.parse(params) : params
    result = db.execute(sql, param_list)
    db.commit rescue nil
    { "success" => true, "affected_rows" => (result.respond_to?(:count) ? result.count : 0) }
  }, "Execute arbitrary SQL (INSERT/UPDATE/DELETE/DDL)")

  server.register_tool("database_tables", lambda {
    db = Tina4.database
    return { "error" => "No database connection" } if db.nil?
    db.tables
  }, "List all database tables")

  server.register_tool("database_columns", lambda { |table:|
    db = Tina4.database
    return { "error" => "No database connection" } if db.nil?
    db.columns(table)
  }, "Get column definitions for a table")

  # ── Route Tools ───────────────────────────────────
  server.register_tool("route_list", lambda {
    routes = Tina4::Router.routes
    routes.map do |route|
      {
        "method"        => route[:method].to_s,
        "path"          => route[:path].to_s,
        "auth_required" => !route[:auth_handler].nil?
      }
    end
  }, "List all registered routes")

  server.register_tool("route_test", lambda { |method:, path:, body: "", headers: "{}"|
    client = Tina4::TestClient.new
    header_hash = headers.is_a?(String) ? JSON.parse(headers) : headers
    m = method.upcase
    r = case m
        when "GET"    then client.get(path, headers: header_hash)
        when "POST"   then client.post(path, body: body, headers: header_hash)
        when "PUT"    then client.put(path, body: body, headers: header_hash)
        when "DELETE" then client.delete(path, headers: header_hash)
        else return { "error" => "Unsupported method: #{method}" }
        end
    { "status" => r.status, "body" => r.body, "content_type" => r.content_type }
  }, "Call a route and return the response")

  server.register_tool("swagger_spec", lambda {
    Tina4::Swagger.generate
  }, "Return the OpenAPI 3.0.3 JSON spec")

  # ── Template Tools ────────────────────────────────
  server.register_tool("template_render", lambda { |template:, data: "{}"|
    ctx = data.is_a?(String) ? JSON.parse(data) : data
    Tina4::Template.render_string(template, ctx)
  }, "Render a template string with data")

  # ── File Tools ────────────────────────────────────
  server.register_tool("file_read", lambda { |path:|
    p = safe_path.call(path)
    return "File not found: #{path}" unless File.exist?(p)
    return "Not a file: #{path}" unless File.file?(p)
    File.read(p, encoding: "utf-8")
  }, "Read a project file")

  server.register_tool("file_write", lambda { |path:, content:|
    p = safe_path.call(path)
    FileUtils.mkdir_p(File.dirname(p))
    File.write(p, content, encoding: "utf-8")
    rel = p.sub("#{project_root}/", "")
    { "written" => rel, "bytes" => content.bytesize }
  }, "Write or update a project file")

  server.register_tool("file_list", lambda { |path: "."|
    p = safe_path.call(path)
    return { "error" => "Directory not found: #{path}" } unless File.exist?(p)
    return { "error" => "Not a directory: #{path}" } unless File.directory?(p)
    Dir.children(p).sort.map do |entry|
      full = File.join(p, entry)
      {
        "name" => entry,
        "type" => File.directory?(full) ? "dir" : "file",
        "size" => File.file?(full) ? File.size(full) : 0
      }
    end
  }, "List files in a directory")

  server.register_tool("asset_upload", lambda { |filename:, content:, encoding: "utf-8"|
    target = safe_path.call("src/public/#{filename}")
    FileUtils.mkdir_p(File.dirname(target))
    if encoding == "base64"
      require "base64"
      File.binwrite(target, Base64.decode64(content))
    else
      File.write(target, content, encoding: "utf-8")
    end
    rel = target.sub("#{project_root}/", "")
    { "uploaded" => rel, "bytes" => File.size(target) }
  }, "Upload a file to src/public/")

  # ── Migration Tools ───────────────────────────────
  server.register_tool("migration_status", lambda {
    db = Tina4.database
    return { "error" => "No database connection" } if db.nil?
    migration = Tina4::Migration.new(db)
    migration.respond_to?(:status) ? migration.status : { "info" => "Migration status not available" }
  }, "List pending and completed migrations")

  server.register_tool("migration_create", lambda { |description:|
    migration = Tina4::Migration.new(nil)
    filename = migration.create(description)
    { "created" => filename }
  }, "Create a new migration file")

  server.register_tool("migration_run", lambda {
    db = Tina4.database
    return { "error" => "No database connection" } if db.nil?
    migration = Tina4::Migration.new(db)
    result = migration.run
    { "result" => result.to_s }
  }, "Run all pending migrations")

  # ── Queue Tools ───────────────────────────────────
  server.register_tool("queue_status", lambda { |topic: "default"|
    begin
      q = Tina4::Queue.new(topic: topic)
      {
        "topic"     => topic,
        "pending"   => q.size("pending"),
        "completed" => q.size("completed"),
        "failed"    => q.size("failed")
      }
    rescue => e
      { "error" => e.message }
    end
  }, "Get queue size by status")

  # ── Session/Cache Tools ───────────────────────────
  server.register_tool("session_list", lambda {
    session_dir = File.join("data", "sessions")
    return [] unless File.directory?(session_dir)
    Dir.glob(File.join(session_dir, "*.json")).map do |f|
      begin
        data = JSON.parse(File.read(f))
        { "id" => File.basename(f, ".json"), "data" => data }
      rescue JSON::ParserError, IOError
        { "id" => File.basename(f, ".json"), "error" => "corrupt" }
      end
    end
  }, "List active sessions")

  server.register_tool("cache_stats", lambda {
    begin
      if defined?(Tina4::ResponseCache)
        cache = Tina4::ResponseCache.new
        cache.cache_stats
      else
        { "error" => "Response cache not available" }
      end
    rescue => e
      { "error" => e.message }
    end
  }, "Get response cache statistics")

  # ── ORM Tools ─────────────────────────────────────
  server.register_tool("orm_describe", lambda {
    models = []
    Tina4::ORM.subclasses.each do |cls|
      fields = cls.field_definitions.map do |name, field|
        {
          "name"        => name.to_s,
          "type"        => field[:type].to_s,
          "primary_key" => field[:primary_key] == true
        }
      end
      models << {
        "class"  => cls.name,
        "table"  => cls.respond_to?(:table_name) ? cls.table_name : cls.name.downcase,
        "fields" => fields
      }
    end
    models
  }, "List all ORM models with fields and types")

  # ── Debugging Tools ───────────────────────────────
  server.register_tool("log_tail", lambda { |lines: 50|
    log_file = File.join("logs", "debug.log")
    return [] unless File.exist?(log_file)
    all_lines = File.read(log_file, encoding: "utf-8").split("\n")
    all_lines.last([lines.to_i, all_lines.length].min)
  }, "Read recent log entries")

  server.register_tool("error_log", lambda { |limit: 20|
    begin
      if defined?(Tina4::DevAdmin) && Tina4::DevAdmin.respond_to?(:message_log)
        log = Tina4::DevAdmin.message_log
        log.respond_to?(:get) ? log.get(category: "error").first(limit.to_i) : []
      else
        []
      end
    rescue
      []
    end
  }, "Recent errors and exceptions")

  server.register_tool("env_list", lambda {
    ENV.sort.to_h { |k, v| [k, redact_env.call(k, v)] }
  }, "List environment variables (secrets redacted)")

  # ── Data Tools ────────────────────────────────────
  server.register_tool("seed_table", lambda { |table:, count: 10|
    begin
      db = Tina4.database
      return { "error" => "No database connection" } if db.nil?
      inserted = Tina4.seed_table(table, db.columns(table), count: count.to_i)
      { "table" => table, "inserted" => inserted }
    rescue => e
      { "error" => e.message }
    end
  }, "Seed a table with fake data")

  # ── System Tools ──────────────────────────────────
  server.register_tool("system_info", lambda {
    {
      "framework" => "tina4-ruby",
      "version"   => (defined?(Tina4::VERSION) ? Tina4::VERSION : "unknown"),
      "ruby"      => RUBY_DESCRIPTION,
      "platform"  => RUBY_PLATFORM,
      "cwd"       => project_root,
      "debug"     => ENV.fetch("TINA4_DEBUG", "false")
    }
  }, "Framework version, Ruby version, project info")
end