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")

  # ── File patch ────────────────────────────────────
  server.register_tool("file_patch", lambda { |path:, old_string:, new_string:, count: 1|
    p = safe_path.call(path)
    return { "error" => "File not found: #{path}" } unless File.file?(p)
    original = File.read(p, encoding: "utf-8")
    occurrences = original.scan(old_string).size
    return { "error" => "old_string not found in #{path}" } if occurrences.zero?
    if occurrences != count.to_i
      return { "error" => "old_string appears #{occurrences} times, expected #{count}. Expand old_string to make it unique, or set count explicitly." }
    end
    updated = original.sub(old_string, new_string)
    # Ruby String#sub replaces first; if count > 1, do N replacements
    if count.to_i > 1
      updated = original.dup
      count.to_i.times { updated.sub!(old_string, new_string) }
    end
    File.write(p, updated, encoding: "utf-8")
    rel = p.sub("#{project_root}/", "")
    Tina4::Plan.record_action("patched", rel) if defined?(Tina4::Plan)
    { "patched" => rel, "replacements" => count.to_i, "bytes" => updated.bytesize }
  }, "Targeted edit: replace old_string with new_string in a file")

  # ── Docs tools ────────────────────────────────────
  framework_doc_paths = lambda do
    gem_root = File.expand_path("..", File.dirname(__FILE__))
    candidates = [
      File.join(gem_root, "..", "CLAUDE.md"),
      File.join(gem_root, "..", "AGENTS.md"),
      File.join(gem_root, "..", "CONVENTIONS.md"),
      File.join(gem_root, "..", "README.md"),
      File.join(Dir.pwd, "CLAUDE.md")
    ]
    candidates.map { |p| File.expand_path(p) }.uniq.select { |p| File.file?(p) }
  end

  server.register_tool("docs_list", lambda {
    framework_doc_paths.call.map { |p| { "name" => File.basename(p), "bytes" => File.size(p) } }
  }, "List framework documentation files")

  server.register_tool("docs_search", lambda { |query:, limit: 5, context_lines: 4|
    return { "error" => "query must be at least 2 characters" } if query.to_s.length < 2
    needle = query.to_s.downcase
    hits = []
    framework_doc_paths.call.each do |p|
      begin
        lines = File.read(p, encoding: "utf-8", invalid: :replace, undef: :replace).split("\n")
      rescue StandardError
        next
      end
      lines.each_with_index do |line, i|
        next unless line.downcase.include?(needle)
        start_i = [0, i - context_lines.to_i].max
        end_i   = [lines.size, i + context_lines.to_i + 1].min
        score = 1
        score += 1 if line.include?(query.to_s)
        score += 2 if line.lstrip.start_with?("#")
        hits << {
          "file" => File.basename(p),
          "line" => i + 1,
          "score" => score,
          "snippet" => lines[start_i...end_i].join("\n")
        }
      end
    end
    hits.sort_by! { |h| -h["score"] }
    hits.first([1, limit.to_i].max)
  }, "Search Tina4 framework docs for a query string")

  server.register_tool("docs_section", lambda { |file:, heading:|
    match = framework_doc_paths.call.find { |p| File.basename(p) == file }
    return { "error" => "Unknown doc file: #{file}. Try docs_list() first." } unless match
    text = File.read(match, encoding: "utf-8", invalid: :replace, undef: :replace)
    lines = text.split("\n")
    heading_lc = heading.to_s.downcase.strip
    start_i = -1
    start_level = 0
    lines.each_with_index do |line, i|
      stripped = line.lstrip
      next unless stripped.start_with?("#")
      level = stripped.length - stripped.sub(/\A#+/, "").length
      title = stripped[level..].to_s.strip.downcase
      if title.include?(heading_lc)
        start_i = i
        start_level = level
        break
      end
    end
    return { "error" => "Heading '#{heading}' not found in #{file}" } if start_i < 0
    end_i = lines.size
    (start_i + 1).upto(lines.size - 1) do |j|
      stripped = lines[j].lstrip
      next unless stripped.start_with?("#")
      level = stripped.length - stripped.sub(/\A#+/, "").length
      if level <= start_level
        end_i = j
        break
      end
    end
    { "file" => file, "heading" => lines[start_i].strip, "body" => lines[start_i...end_i].join("\n") }
  }, "Return a full markdown section from a framework doc file")

  # ── Git / deps / project ──────────────────────────
  server.register_tool("git_status", lambda {
    Tina4::DevAdmin.send(:git_status_payload)
  }, "Show git branch, modified/untracked files, recent commits")

  server.register_tool("deps_list", lambda {
    gemfile = File.join(Dir.pwd, "Gemfile")
    return { "error" => "No Gemfile at project root" } unless File.file?(gemfile)
    deps = File.read(gemfile).scan(/^\s*gem\s+["']([^"']+)["']/).flatten
    { "name" => File.basename(Dir.pwd), "dependencies" => deps }
  }, "List this project's declared Ruby dependencies")

  server.register_tool("project_overview", lambda {
    { "system" => { "framework" => "tina4-ruby", "version" => (defined?(Tina4::VERSION) ? Tina4::VERSION : "unknown"), "ruby" => RUBY_DESCRIPTION, "cwd" => project_root } }
  }, "One-shot snapshot: system + project info")

  # ── Project index ─────────────────────────────────
  server.register_tool("index_rebuild", lambda {
    Tina4::ProjectIndex.refresh
  }, "Refresh the persistent project index (lazy, mtime-based)")

  server.register_tool("index_search", lambda { |query:, limit: 20|
    Tina4::ProjectIndex.search(query, limit.to_i)
  }, "Find files by path, symbol, route, or summary")

  server.register_tool("index_file", lambda { |path:|
    Tina4::ProjectIndex.file_entry(path)
  }, "Full index entry for one file")

  server.register_tool("index_overview", lambda {
    Tina4::ProjectIndex.overview
  }, "Project shape: files by language, routes, models, recent edits")

  # ── Plan management ───────────────────────────────
  server.register_tool("plan_current", lambda {
    Tina4::Plan.current
  }, "The active plan: title, steps (done/not), next step, progress")

  server.register_tool("plan_list", lambda {
    Tina4::Plan.list_plans
  }, "All plans in plan/ with progress and which one is active")

  server.register_tool("plan_create", lambda { |title:, goal: "", steps: nil, make_current: true|
    Tina4::Plan.create(title, goal: goal, steps: steps, make_current: make_current)
  }, "Create a new markdown plan in plan/ and make it active")

  server.register_tool("plan_switch_to", lambda { |name:|
    Tina4::Plan.set_current(name)
  }, "Make a different plan the active one")

  server.register_tool("plan_complete_step", lambda { |index:|
    Tina4::Plan.complete_step(index.to_i)
  }, "Tick a step as done (call the moment the step finishes)")

  server.register_tool("plan_add_step", lambda { |text:|
    Tina4::Plan.add_step(text)
  }, "Append a new unchecked step to the current plan")

  server.register_tool("plan_note", lambda { |text:|
    Tina4::Plan.append_note(text)
  }, "Append a timestamped note/breadcrumb to the current plan")

  server.register_tool("plan_archive", lambda { |name: ""|
    Tina4::Plan.archive(name)
  }, "Move a finished plan to plan/done/")

  server.register_tool("plan_read", lambda { |name:|
    Tina4::Plan.read(name)
  }, "Full structured view of any plan by filename")

  server.register_tool("plan_flesh", lambda { |name: "", prompt: ""|
    Tina4::Plan.flesh(name, prompt)
  }, "Auto-generate concrete build steps via AI and append them to an existing plan")

  # ── 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