Chapter 28: Building Custom MCP Servers#

require "tina4"​mcp = Tina4::McpServer.new("/api/my-tools", name: "My App Tools", version: "1.0.0")

mcp.register_routes

require "tina4"​mcp = Tina4::McpServer.new("/crm/mcp", name: "CRM Tools")​Tina4.mcp_tool("lookup_customer", description: "Find a customer by email", server: mcp,  params: [{ name: "email", type: "string" }]) do |args|  db.fetch_one("SELECT * FROM customers WHERE email = ?", [args["email"]])end​Tina4.mcp_tool("recent_orders", description: "Get recent orders for a customer", server: mcp,  params: [    { name: "customer_id", type: "integer" },    { name: "limit", type: "integer", default: 10 }  ]) do |args|  db.fetch(    "SELECT * FROM orders WHERE customer_id = ? ORDER BY created_at DESC",    [args["customer_id"]], limit: args["limit"] || 10  ).to_aend

{  "name": "lookup_customer",  "description": "Find a customer by email",  "inputSchema": {    "type": "object",    "properties": {      "email": {"type": "string"}    },    "required": ["email"]  }}

Tina4.mcp_resource("crm://product-catalog", description: "All active products", server: mcp) do  db.fetch("SELECT id, name, price, category FROM products WHERE active = 1").to_aend​Tina4.mcp_resource("crm://tax-rates", description: "Current tax rates by region", server: mcp) do  db.fetch("SELECT region, rate FROM tax_rates").to_aend

require "tina4"​mcp = Tina4::McpServer.new("/accounting/mcp", name: "Accounting Tools")​class AccountingService  def initialize(db)    @db = db  end​  def lookup(invoice_no)    @db.fetch_one("SELECT * FROM invoices WHERE invoice_no = ?", [invoice_no])  end​  def balances(min_amount = 0.0)    @db.fetch("SELECT * FROM invoices WHERE paid = 0 AND total >= ?", [min_amount]).to_a  end​  def summary(year, month)    @db.fetch_one(      "SELECT SUM(total) as revenue, COUNT(*) as invoice_count " \      "FROM invoices WHERE strftime('%Y', created_at) = ? " \      "AND strftime('%m', created_at) = ?",      [year.to_s, format("%02d", month)]    )  endend​svc = AccountingService.new(db)​Tina4.mcp_tool("invoice_lookup", description: "Find an invoice by number", server: mcp,  params: [{ name: "invoice_no", type: "string" }]) do |args|  svc.lookup(args["invoice_no"])end​Tina4.mcp_tool("outstanding_balances", description: "List all unpaid invoices", server: mcp,  params: [{ name: "min_amount", type: "number", default: 0.0 }]) do |args|  svc.balances(args["min_amount"] || 0.0)end​Tina4.mcp_tool("monthly_summary", description: "Revenue summary for a month", server: mcp,  params: [    { name: "year", type: "integer" },    { name: "month", type: "integer" }  ]) do |args|  svc.summary(args["year"], args["month"])end

mcp.register_routes​# Require a valid bearer token on every request under the MCP pathTina4::Middleware.before("#{mcp.path}/*") do |request, response|  token = request.headers["Authorization"].to_s.sub(/^Bearer /, "")  unless Tina4::Auth.valid_token(token)    response.json({ error: "Unauthorized" }, 401)  endend

Tina4.mcp_tool("sensitive_data", description: "Access restricted data", server: mcp,  params: [{ name: "token", type: "string" }]) do |args|  payload = Tina4::Auth.valid_token(args["token"])  unless payload    next { error: "Unauthorized" }  end  db.fetch("SELECT * FROM sensitive_table").to_aend

# Test the tool block directlydef test_lookup_customer  result = db.fetch_one("SELECT * FROM customers WHERE email = ?", ["alice@example.com"])  assert result != nil  assert_equal "alice@example.com", result["email"]end​# Test via MCP protocoldef test_mcp_tool_call  resp = JSON.parse(mcp.handle_message({    "jsonrpc" => "2.0",    "id" => 1,    "method" => "tools/call",    "params" => {      "name" => "lookup_customer",      "arguments" => { "email" => "alice@example.com" }    }  }))  assert resp.key?("result")  content = resp["result"]["content"][0]["text"]  assert content.downcase.include?("alice")end

# app.rbrequire "tina4"​db = Tina4::Database.new("sqlite:///crm.db")Tina4.bind_database(db)   # models resolve Tina4.database; or set per-model with Model.db = db​# Create MCP servercrm_mcp = Tina4::McpServer.new("/crm/mcp", name: "CRM Assistant", version: "1.0.0")​# ToolsTina4.mcp_tool("find_customer", description: "Search customers by name or email", server: crm_mcp,  params: [{ name: "query", type: "string" }]) do |args|  q = args["query"]  db.fetch("SELECT * FROM customers WHERE name LIKE ? OR email LIKE ?",           ["%#{q}%", "%#{q}%"]).to_aend​Tina4.mcp_tool("customer_orders", description: "Get all orders for a customer", server: crm_mcp,  params: [{ name: "customer_id", type: "integer" }]) do |args|  db.fetch(    "SELECT o.*, GROUP_CONCAT(oi.product_name) as items " \    "FROM orders o LEFT JOIN order_items oi ON o.id = oi.order_id " \    "WHERE o.customer_id = ? GROUP BY o.id ORDER BY o.created_at DESC",    [args["customer_id"]]  ).to_aend​Tina4.mcp_tool("create_note", description: "Add a note to a customer record", server: crm_mcp,  params: [    { name: "customer_id", type: "integer" },    { name: "note", type: "string" }  ]) do |args|  db.insert("customer_notes", { customer_id: args["customer_id"], note: args["note"] })  { success: true }end​# ResourcesTina4.mcp_resource("crm://products", description: "Product catalog", server: crm_mcp) do  db.fetch("SELECT * FROM products WHERE active = 1").to_aend​Tina4.mcp_resource("crm://stats", description: "CRM statistics", server: crm_mcp) do  customers = db.fetch_one("SELECT COUNT(*) as count FROM customers")  orders = db.fetch_one("SELECT COUNT(*) as count, SUM(total) as revenue FROM orders")  {    customers: customers["count"],    orders: orders["count"],    revenue: orders["revenue"]  }end​# Register routescrm_mcp.register_routes​Tina4.run