← 返回 Server 节点通过管道和流量控制阀连接到圆柱形数据库的蓝图线条图,配有压力表

在 TypeScript 上的 Supabase 中构建 MCP Server

MCP TypeScript SDK 与 Supabase Edge Functions 相结合,为你提供一个托管的、全球分布的 server,它位于 AI 客户端和数据库之间。AI 客户端无法获得原始数据库访问权限。根据 Supabase 部署指南,此堆栈推荐的传输方式是 WebStandardStreamableHTTPServerTransport。以下仅涵盖此实现。你将获得:一个合成的只读产品目录、分页查询工具、RLS 强制的双用户隔离、Inspector 记录演示,以及部署和版本控制的注意事项。

Server 和数据流

MCP 位于 AI 客户端(Claude Code、Claude Desktop、Cursor)和你的后端之间。客户端通过 HTTP 向你的 server 发送 JSON-RPC。你的 server 公开工具。工具是客户端可以调用的类型化函数。Server 决定这些工具可以访问什么。

对于 Supabase,这个边界至关重要。正如 UI Bakery 的分析所述:AI 客户端不应该获得对数据库的无限直接访问权限。你设计工具,定义查询,Supabase 行级安全(RLS)根据调用者的身份处理逐行访问。

数据流看起来像这样:

  1. Claude Code 向你的 Edge Function URL 发送 JSON-RPC tools/call 请求。
  2. Edge Function 接收它,从 Authorization header 中提取 JWT,并创建一个限定到该身份的 Supabase 客户端。
  3. 工具运行查询。表上的 RLS 策略过滤调用者可以看到的行。
  4. 结果序列化为 JSON-RPC 响应返回。

注意这个链中没有 service-role key。Service-role 客户端完全绕过 RLS,所以如果你在面向调用者的工具中使用它,你的 RLS 策略就形同虚设。Service-role key 仅用于后台管理任务。

合成样本数据

这里的示例使用一个有三列的合成 products 表:id(uuid)、owner_id(uuid,auth.users 的外键)和 name(text)。两个合成用户 alice bob,各自拥有不相交的行集合。此设置纯粹用于说明目的,不基于实际客户项目。

create table public.products (

id uuid primary key default gen_random_uuid(),

owner_id uuid references auth.users(id),

name text not null

);

使用各自的 UUID 为每个用户生成约十行数据。

使用分页定义小工具界面

工具越窄越好。返回整个表的工具是一个隐患。分页使 payloads 可预测,避免触及 Edge Function 响应限制。

此 server 的工具界面刻意很小:

  • list_products:返回调用者拥有的产品页面,接受 page page_size 参数,默认为第 1 页,每页 20 行。
  • get_product:按 id 返回单个产品,如果调用者不拥有它则优雅失败。

就这样。两个工具。你不是在构建查询引擎。如果你想了解如何在 Supabase 上构建更大的目录式读取层,值得先阅读那篇 25,000 页的目录文章。

用Zod定义schema

MCP TypeScript SDK使用Zod进行输入验证。你的工具定义大致如下(示意用途):

import { z } from 'npm:zod'

``

const ListProductsInput = z.object({

page: z.number().int().min(1).default(1),

page_size: z.number().int().min(1).max(100).default(20),

})

``

const GetProductInput = z.object({

id: z.string().uuid(),

})

Zod在你的处理器运行前进行验证。如果客户端发送无效数据,客户端会收到一个类型化的错误,而不是运行时异常。这是从一开始就应该建立的良好习惯。

实现并运行TypeScript服务器

Supabase部署指南使用@modelcontextprotocol/sdk@1.25.3和WebStandardStreamableHTTPServerTransport。这是撰写时锁定的版本。在你发布前检查最新文档。

脚架搭建

mkdir my-mcp-server && cd my-mcp-server

supabase init

supabase functions new mcp

你的函数位于supabase/functions/mcp/index.ts。用类似下面的内容替换默认代码(示意结构):

import 'jsr:@supabase/functions-js/edge-runtime.d.ts'

import { McpServer } from 'npm:@modelcontextprotocol/sdk@1.25.3/server/mcp.js'

import { WebStandardStreamableHTTPServerTransport } from 'npm:@modelcontextprotocol/sdk@1.25.3/server/streamableHttp.js'

import { createClient } from 'npm:@supabase/supabase-js@2'

import { z } from 'npm:zod'

``

Deno.serve(async (req) => {

const authHeader = req.headers.get('Authorization') ?? ''

const jwt = authHeader.replace('Bearer ', '')

``

const supabase = createClient(

Deno.env.get('SUPABASE_URL')!,

Deno.env.get('SUPABASE_ANON_KEY')!,

{ global: { headers: { Authorization: Bearer ${jwt} } } }

)

``

const server = new McpServer({ name: 'products-mcp', version: '1.0.0' })

``

server.tool('list_products', 'List products owned by the caller',

{ page: z.number().int().min(1).default(1), page_size: z.number().int().min(1).max(100).default(20) },

async ({ page, page_size }) => {

const from = (page - 1) * page_size

const { data, error } = await supabase

.from('products')

.select('id, name')

.range(from, from + page_size - 1)

if (error) return { content: [{ type: 'text', text: Error: ${error.message} }] }

return { content: [{ type: 'text', text: JSON.stringify(data) }] }

}

)

``

server.tool('get_product', 'Get a single product by id',

{ id: z.string().uuid() },

async ({ id }) => {

const { data, error } = await supabase

.from('products')

.select('id, name')

.eq('id', id)

.maybeSingle()

if (error) return { content: [{ type: 'text', text: Error: ${error.message} }] }

if (!data) return { content: [{ type: 'text', text: 'Not found or access denied' }] }

return { content: [{ type: 'text', text: JSON.stringify(data) }] }

}

)

``

const transport = new WebStandardStreamableHTTPServerTransport({ path: '/mcp' })

await server.connect(transport)

return transport.handleRequest(req)

})

这里有两点值得注意。首先,每个请求创建一个新的McpServer和transport实例是Edge Function运行时期望的无状态模式。其次,Supabase客户端使用SUPABASE_ANON_KEY而不是service-role密钥创建,JWT在请求头中转发以便RLS能识别调用方的身份。

本地运行

supabase start

supabase functions serve --no-verify-jwt mcp

你的服务器在http://localhost:54321/functions/v1/mcp上可用。--no-verify-jwt标志对本地测试没问题。生产环境中你需要启用JWT验证。

验证调用方并强制行级访问控制

大多数教程在这里出错。他们使用service-role密钥是因为它更简单。但这绕过了RLS,意味着你的工具表面,无论多么狭窄,都拥有对每一行的管理员级读取权限。这对面向调用方的服务器来说是不可接受的。

蓝图线描:由过滤门分隔的同心数据库圆柱体,带有显示行级访问分区的方向流箭头

上面显示的模式(传递调用方的JWT,使用anon密钥)意味着Supabase Auth评估令牌并在Postgres内为每个查询设置auth.uid()上下文。你的RLS策略可以这样使用它:

alter table public.products enable row level security;

``

create policy "owners can read own products"

on public.products

for select

using (owner_id = auth.uid());

现在alice的JWT只能返回owner_id匹配她的UID的行。bob调用get_product并使用Alice产品ID之一会得到"Not found or access denied",不是错误,也不是Alice的数据。这是正确的行为。

若要深入了解编写RLS策略,Supabase RLS指南详细讲解了常见模式和陷阱。

访问隔离检查(示意用途)

两用户检查很直接。通过supabase.auth.signInWithPassword获取Alice的JWT,调用list_products,确认所有返回的owner_id值与Alice的UID匹配。用Bob的JWT重复。确认零交叉污染。如果你的策略正确,这按构造通过。如果你犯了service-role错误,这里就会暴露,因为两个用户都会看到所有行。

如果你的Supabase项目已经在处理复杂的Next.js数据获取需求和这个功能,/solutions/nextjs-supabase-development/的团队可以帮你构造数据库层,而不用会产生这些问题的service-role快捷方式。

连接Claude Code并验证错误情况

在本地运行函数后,将其添加到 Claude Code:

claude mcp add products-mcp -t http http://localhost:54321/functions/v1/mcp

然后,从 Claude Code 会话中,你可以直接调用工具。你应该在可用工具列表中看到 list_products 和 get_product。如果没有,运行 claude mcp list 以确认服务器已正确注册。

Inspector 脚本演练

MCP Inspector(npx @modelcontextprotocol/inspector)为你提供浏览器 UI 来测试工具,无需完整客户端。如 mcp-lite 指南所述,你将端点 URL 粘贴到 Inspector 中,它会自动发现你的工具。

list_products 的示例 Inspector 会话:

Request:

{

"jsonrpc": "2.0",

"method": "tools/call",

"params": {

"name": "list_products",

"arguments": { "page": 1, "page_size": 5 }

},

"id": 1

}

``

Response (Alice's JWT):

{

"result": {

"content": [{ "type": "text", "text": "[{\"id\":\"...\",\"name\":\"Widget A\"}...]" }]

}

}

现在有意地测试错误情况。使用 Alice 的 JWT 调用 get_product,传入属于 Bob 的有效 UUID。你应该获得 Not found or access denied。使用 page_size: 200(超过 100 的上限)调用 list_products。Zod 应该在查询运行前拒绝它。两者都是预期行为,部署前值得确认。

JWT 缺失时会发生什么?

没有 Authorization 标头,你的 createClient 调用会传递空 JWT。Supabase 将其视为 anon 角色。如果你的 RLS 策略未明确授予 anon 角色访问权限,查询返回零行。这是正确的。工具返回空数组,不是错误。你可能倾向于返回明确的"未认证"消息。无论哪种方式,都要有意决策,而不是在生产环境中发现。

部署、观察和版本化服务器

部署是一条命令:

supabase functions deploy mcp

你的函数获得稳定 URL:https://<project-ref>.supabase.co/functions/v1/mcp。在 Supabase 仪表盘中设置环境变量(设置 > Edge Functions > Secrets):SUPABASE_URL 和 SUPABASE_ANON_KEY。函数运行时在调用时注入这些变量。

可观测性

Supabase 在仪表盘中提供内置 Edge Function 日志。按函数名称筛选。你将看到冷启动时间、执行时长和任何未捕获的错误。为获得更结构化的可观测性,从工具处理程序内部写入 JSON 格式的 console.log。日志流会拾取它。

对于生产使用,考虑:

  • 为每个请求记录工具名称和调用者 UID(不是完整 JWT)。
  • 对 Supabase 查询设置超时,防止慢查询消耗 Edge Function 执行预算。
  • 返回确定性错误形状,以便 AI 客户端可以一致地处理失败。

版本化

Edge Functions 没有内置版本控制。实用方法是基于路径的:将 v2 函数部署为单独的 Edge Function(supabase functions new mcp-v2),独立测试,然后更新你的 Claude Code 注册。旧客户端可以继续指向 v1 URL,直到你切换它们。

如果你正在构建具有多个服务器的生产 MCP 堆栈,production MCP stack 文章涵盖了超出本范围的服务器选择和架构决策。

部署清单

  1. 部署前,从本地 serve 命令中移除 --no-verify-jwt
  2. 确认 SUPABASE_ANON_KEY 在仪表盘机密中设置,而不是 service-role。
  3. 在宣布端点前,针对生产 URL 运行两用户访问隔离检查。
  4. 在每个工具中设置 max_rows 上限,防止无界查询。
  5. 检查导入中固定的 SDK 版本(Supabase 文档撰写时为 @modelcontextprotocol/sdk@1.25.3)是否与当前版本一致,如需要则更新。

FAQ

我可以使用 mcp-lite 代替官方 SDK 吗?

可以。Supabase 文档明确指出你可以使用 mcp-lite 或 mcp-handler 作为官方 SDK 的替代方案。上面使用的 WebStandardStreamableHTTPServerTransport 来自官方 SDK,但 mcp-lite 更轻量且零依赖。选择其一并保持一致。mcp-lite 指南展示了脚手架命令:npm create mcp-lite@latest。上述架构和 RLS 行为与你选择的框架无关。

部署 Edge Functions 需要付费的 Supabase 计划吗?

Edge Functions 在免费层可用,但在调用次数和执行时间上有限制。查看当前 Supabase 定价页了解具体数字。对于低流量的内部工具,免费层通常足够。对于任何面向生产且有实际查询量的服务,Pro 计划的限制更适合。

如果我的 MCP 服务器需要写入数据而不仅仅是读取呢?

添加运行 insert、update 或 delete 查询的工具。RLS 模式仍然适用:你的策略根据 auth.uid() 控制调用者可以修改的行。主要的额外考虑是幂等性。如果 AI 客户端重试失败的工具调用,你是否希望 insert 运行两次?使用具有稳定主键的 upsert 或在插入前检查存在性是标准的缓解方案。

没有 Supabase 账户可以在本地运行吗?

可以,supabase start 通过 Docker 在本地运行完整的 Supabase 堆栈,包括 Postgres、Auth 和 Edge Functions。你会从 CLI 输出中获得本地 SUPABASE_URL SUPABASE_ANON_KEY。你的函数和 RLS 策略的工作方式与生产环境完全相同。唯一的区别是 JWT 发行者,本地使用 GoTrue 实例而非 Supabase 的云 Auth。

如何处理工具的速率限制?

Edge Functions 没有内置的按调用者速率限制。实用选项有:一个 Postgres 表记录每个 UID 在时间窗口内的调用次数(在工具内检查并递增),或一个上游 API 网关(Supabase 自身的 API 网关在项目级别处理其中一些)。对于大多数内部工具,首先遇到的约束是免费层项目的 Edge Function 调用限制,而不是故意滥用。

上述所有内容中最尖锐的警告:如果在面向调用者的工具中使用 service-role 密钥,你的 RLS 策略将毫无作用。anon 密钥加上 JWT 转发不是锦上添花,它是实现调用者隔离的机制。

← 返回