Skip to main content

7 posts tagged with "saas-development"

View all tags

Full API Mocking with Playwright page.route()

· 4 min read

Encountered this issue while building an AI Agent platform. Here's the root cause and solution.

TL;DR

Use page.route() to intercept all API requests and return predefined mock data. Tests don't depend on real backend, can run stably in any environment, and avoid side effects like creating/deleting data.

Problem

E2E tests calling real API:

// ❌ Depends on real backend
test('create agent', async ({ authenticatedPage }) => {
await authenticatedPage.goto('/')

// Click create button
await authenticatedPage.click('button:has-text("Create")')

// Fill form
await authenticatedPage.fill('input[name="name"]', 'Test Agent')
await authenticatedPage.click('button[type="submit"]')

// Wait for API response
await authenticatedPage.waitForTimeout(2000)

// Verify... but what if backend is down? DB connection failed?
})

Problems:

  1. Depends on backend state — Tests fail when backend is down
  2. Data side effects — Each run creates real data
  3. Not repeatable — Data changes cause assertion failures
  4. CI environment issues — Need full backend service running

Solution

1. Define Mock Data

// e2e/fixtures.ts
export const mockAgents = [
{
agent_id: 'agent-1',
user_id: 'test-user-id',
name: 'Test Agent 1',
skills: [],
mcp_tools: [],
llm_config: { provider: 'anthropic', model: 'claude-3-5-sonnet-20241022' },
risk_threshold: 'medium',
auto_confirm_low: true,
created_at: '2024-01-01T00:00:00Z',
},
]

export const mockSkills = [
{
skill_id: 'skill-1',
owner_id: 'test-user-id',
name: 'Test Skill',
system_prompt: 'You are helpful.',
is_public: false,
is_own: true,
},
]

2. Setup API Mock

export async function setupMockApi(page: Page) {
// Mock GET /api/agents
await page.route('**/api/agents', async (route) => {
if (route.request().method() === 'GET') {
await route.fulfill({
status: 200,
contentType: 'application/json',
body: JSON.stringify(mockAgents),
})
} else if (route.request().method() === 'POST') {
// Simulate creation
const body = route.request().postDataJSON()
const newAgent = {
agent_id: `agent-${Date.now()}`,
user_id: 'test-user-id',
name: body.name,
created_at: new Date().toISOString(),
...body,
}
mockAgents.push(newAgent)
await route.fulfill({
status: 201,
contentType: 'application/json',
body: JSON.stringify(newAgent),
})
}
})

// Mock GET/PATCH/DELETE /api/agents/:id
await page.route('**/api/agents/*', async (route) => {
const url = route.request().url()
const match = url.match(/\/api\/agents\/([^/]+)/)
const agentId = match?.[1]

if (route.request().method() === 'GET') {
const agent = mockAgents.find((a) => a.agent_id === agentId)
await route.fulfill({
status: agent ? 200 : 404,
contentType: 'application/json',
body: JSON.stringify(agent || { error: 'Not found' }),
})
} else if (route.request().method() === 'DELETE') {
const index = mockAgents.findIndex((a) => a.agent_id === agentId)
if (index !== -1) mockAgents.splice(index, 1)
await route.fulfill({ status: 204 })
}
})

// Mock /api/skills
await page.route('**/api/skills**', async (route) => {
await route.fulfill({
status: 200,
contentType: 'application/json',
body: JSON.stringify(mockSkills),
})
})

// Mock /api/health
await page.route('**/api/health', async (route) => {
await route.fulfill({
status: 200,
contentType: 'application/json',
body: JSON.stringify({ status: 'ok' }),
})
})
}

3. Use in Fixture

export const test = base.extend<{ authenticatedPage: Page }>({
authenticatedPage: async ({ page }, use) => {
await setupMockAuth(page)
await setupMockApi(page) // Intercept all API
await use(page)
},
})

4. Use in Tests

// e2e/dashboard.spec.ts
import { test, expect, mockAgents } from './fixtures'

test('should display agent list', async ({ authenticatedPage }) => {
await authenticatedPage.goto('/')

// API automatically mocked, returns mockAgents
const agentCards = authenticatedPage.locator('[data-testid="agent-card"]')

// Assertion based on known mock data
await expect(agentCards).toHaveCount(mockAgents.length)
})

test('should create new agent', async ({ authenticatedPage }) => {
await authenticatedPage.goto('/')

await authenticatedPage.click('button:has-text("Create")')

const dialog = authenticatedPage.locator('[role="dialog"]')
await dialog.locator('input[name="name"]').fill('New Agent')
await dialog.locator('button[type="submit"]').click()

// POST /api/agents mocked, returns 201
await authenticatedPage.waitForTimeout(500)

// Verify UI update
await expect(authenticatedPage.locator('text=New Agent')).toBeVisible()
})

Core Techniques

URL Matching Patterns

// Exact match
await page.route('**/api/agents', handler)

// Wildcard match
await page.route('**/api/agents/**', handler)

// Regex match
await page.route(/\/api\/agents\/\d+/, handler)

Reading Request Body

await page.route('**/api/agents', async (route) => {
const body = route.request().postDataJSON()
console.log('Request body:', body)

// Return different responses based on request content
if (body.name === 'error-test') {
await route.fulfill({ status: 400, body: JSON.stringify({ error: 'Bad request' }) })
} else {
await route.fulfill({ status: 201, body: JSON.stringify({ id: 'new-id', ...body }) })
}
})

Simulating Error Scenarios

// Network error
await page.route('**/api/agents', (route) => route.abort('failed'))

// Timeout
await page.route('**/api/agents', async (route) => {
await new Promise((r) => setTimeout(r, 30000))
route.continue()
})

// 500 error
await page.route('**/api/agents', (route) =>
route.fulfill({ status: 500, body: JSON.stringify({ error: 'Internal error' }) })
)

Partial Pass-through

// Only mock specific API, pass through others
await page.route('**/api/**', async (route) => {
const url = route.request().url()

if (url.includes('/api/agents')) {
await route.fulfill({ status: 200, body: JSON.stringify(mockAgents) })
} else {
await route.continue() // Other APIs use real requests
}
})

Mock Data Management

// Centralize all mock data
// e2e/fixtures.ts
export const mockData = {
agents: [...],
skills: [...],
apiKeys: [...],
tasks: [...],
}

// Reset before each test
test.beforeEach(() => {
Object.assign(mockData, JSON.parse(JSON.stringify(originalMockData)))
})

Key Principles

  1. Mock all external dependencies — API, OAuth, third-party services
  2. Match real data structures — Mock data should match API contract
  3. Cover success and failure — Test 200/400/500 scenarios
  4. Isolate test data — Each test uses independent mock data copy

Interested in similar solutions? Contact us

Skip Login in Playwright Tests with Custom Fixtures

· 3 min read

Encountered this issue while building an AI Agent platform. Here's the root cause and solution.

TL;DR

Use Playwright's test.extend() to create a custom fixture that injects auth token into localStorage via page.addInitScript() before page load. Tests use authenticatedPage instead of page, automatically getting logged-in state without repeating login in each test.

Problem

E2E tests need to verify pages behind authentication:

// ❌ Every test goes through login
test('dashboard', async ({ page }) => {
await page.goto('/login')
await page.fill('input[name="email"]', 'test@example.com')
await page.fill('input[name="password"]', 'password')
await page.click('button[type="submit"]')
await page.waitForURL('/dashboard')
// Finally can start testing...
await expect(page.locator('h1')).toBeVisible()
})

Problems:

  1. Repeated login in every test — Wastes time, slows CI
  2. Depends on real auth service — Tests fail when Supabase Auth is unavailable
  3. State pollution between tests — Login state may affect other tests

Solution

1. Create Custom Fixture

// e2e/fixtures.ts
import { test as base, expect, Page } from '@playwright/test'

// Mock user data
export const mockUser = {
id: 'test-user-id',
email: 'test@example.com',
aud: 'authenticated',
role: 'authenticated',
}

export const mockAccessToken = 'mock-access-token-for-testing'

// Inject auth token into localStorage
export async function setupMockAuth(page: Page) {
await page.addInitScript(
({ user, accessToken }) => {
const mockSession = {
access_token: accessToken,
refresh_token: 'mock-refresh-token',
expires_in: 3600,
expires_at: Math.floor(Date.now() / 1000) + 3600,
token_type: 'bearer',
user,
}
// Supabase auth token key format: sb-{project}-auth-token
localStorage.setItem('sb-placeholder-auth-token', JSON.stringify(mockSession))
},
{ user: mockUser, accessToken: mockAccessToken }
)
}

// Extend fixture
export const test = base.extend<{
authenticatedPage: Page
}>({
authenticatedPage: async ({ page }, use) => {
await setupMockAuth(page)
await use(page)
},
})

export { expect }

2. Use authenticatedPage in Tests

// e2e/dashboard.spec.ts
import { test, expect } from './fixtures'

test.describe('Dashboard', () => {
test('should display welcome message', async ({ authenticatedPage }) => {
// Direct access to protected page, no login needed
await authenticatedPage.goto('/dashboard')

await expect(authenticatedPage.locator('h1')).toContainText('Welcome')
})

test('should show agent list', async ({ authenticatedPage }) => {
await authenticatedPage.goto('/dashboard')

const agents = authenticatedPage.locator('[data-testid="agent-card"]')
await expect(agents.first()).toBeVisible()
})
})

3. Comparison: Unauthenticated vs Authenticated

// Unauthenticated test (redirects to login)
test('redirects to login when not authenticated', async ({ page }) => {
await page.goto('/dashboard')
await page.waitForTimeout(500)

expect(page.url()).toContain('/login')
})

// Authenticated test (direct dashboard access)
test('allows access when authenticated', async ({ authenticatedPage }) => {
await authenticatedPage.goto('/dashboard')
await authenticatedPage.waitForTimeout(500)

expect(authenticatedPage.url()).not.toContain('/login')
})

Core Principles

addInitScript vs Setting After Page Load

// ❌ Wrong: Set after page load (may have already redirected)
await page.goto('/dashboard')
await page.evaluate(() => {
localStorage.setItem('auth-token', '...')
})
// Auth guard already detected unauthenticated and redirected

// ✅ Correct: Inject before page load
await page.addInitScript(() => {
localStorage.setItem('auth-token', '...')
})
await page.goto('/dashboard') // Auth guard detects token on load

addInitScript executes:

  1. Before page DOM parsing
  2. Before React/Vue framework initialization
  3. Before auth guard checks

Fixture Lifecycle

test('dashboard', async ({ authenticatedPage }) => {})

base.extend<{ authenticatedPage }>()

authenticatedPage: async ({ page }, use) => {
await setupMockAuth(page) // 1. Setup auth
await use(page) // 2. Run test
} // 3. Auto cleanup

Complete Configuration

// playwright.config.ts
export default defineConfig({
testDir: './e2e',
fullyParallel: false, // Serial execution to avoid state pollution
workers: 1,
use: {
baseURL: 'http://localhost:5173',
trace: 'on-first-retry',
},
webServer: {
command: 'pnpm dev',
url: 'http://localhost:5173',
reuseExistingServer: !process.env.CI,
},
})

Extension: Combined with API Mock

export const test = base.extend<{
authenticatedPage: Page
}>({
authenticatedPage: async ({ page }, use) => {
await setupMockAuth(page)
await setupMockApi(page) // Also mock API
await use(page)
},
})

Key Principles

  1. Fixture over beforeEach — Automatic reuse, cleaner code
  2. addInitScript avoids race conditions — Inject token before auth guard checks
  3. Isolate test data — Don't mix mock users/tokens with production
  4. Serial execution prevents pollutionworkers: 1 or fullyParallel: false

Interested in similar solutions? Contact us

Bypass Supabase Auth for Playwright E2E Testing Without Login

· 4 min read

Encountered this issue while building an AI Agent SaaS platform for a client. Here's the root cause and solution.

TL;DR

E2E tests shouldn't depend on real OAuth login flows. By detecting localStorage test markers in the useAuth hook, you can inject mock auth state directly and skip Supabase initialization. Also change the Zustand store's loading default to false to prevent AuthGuard from showing an infinite spinner.

Problem

When testing a React SPA with Playwright, pages are protected by AuthGuard and require Supabase authentication. After the test starts, the page shows a loading spinner indefinitely and never reaches the business logic.

// AuthGuard component - tests get stuck here
export function AuthGuard({ children }: AuthGuardProps) {
const { isAuthenticated, loading } = useAuth()

if (loading) {
return <Spinner /> // Forever showing spinner
}

if (!isAuthenticated) {
return <Navigate to="/login" />
}

return <>{children}</>
}

Test code tries to simulate login, but Supabase Auth SDK's internal state can't be controlled by simple API mocking.

Root Cause

1. Supabase Auth Initializes Asynchronously

The useAuth hook calls supabase.auth.getSession() in useEffect, which is async. In test environments, network requests may fail or timeout, leaving state stuck at loading: true.

2. Zustand Store Default Value Problem

// authStore.ts - problematic code
export const useAuthStore = create<AuthState>()(
persist(
(set) => ({
user: null,
token: null,
loading: true, // 👈 Default is true
// ...
}),
{ name: 'auth-storage' }
)
)

At test startup: loading: true + async init failure = forever loading.

3. OAuth Flow Can't Be Automated

Even if APIs can be mocked, OAuth redirect flows require real browser interaction that E2E tests can't reliably simulate.

Solution

Step 1: Add Test Mode Detection in useAuth Hook

// hooks/useAuth.ts
export function useAuth() {
const { user, token, loading, setUser, setToken, setLoading } = useAuthStore()

useEffect(() => {
const initAuth = async () => {
// 👇 Check test mode first
const testAuthUser = localStorage.getItem('test-auth-user')
const testAuthToken = localStorage.getItem('test-auth-token')

if (testAuthUser && testAuthToken) {
try {
const userData = JSON.parse(testAuthUser) as User
setUser(userData)
setToken(testAuthToken)
setLoading(false)
console.log('[useAuth] Using test mode auth')
return // 👈 Return early, skip Supabase init
} catch (e) {
console.error('Failed to parse test auth user:', e)
}
}

// 👇 Normal mode: proceed with Supabase Auth
try {
const { data: { session } } = await supabase.auth.getSession()
if (session) {
setUser(session.user as User)
setToken(session.access_token)
}
} catch (error) {
console.error('Auth init failed:', error)
} finally {
setLoading(false)
}
}

initAuth()

// 👇 Skip auth state listener in test mode
if (localStorage.getItem('test-auth-user')) {
return
}

const { data: { subscription } } = supabase.auth.onAuthStateChange(
async (event, session) => {
// ... normal auth state handling
}
)

return () => subscription.unsubscribe()
}, [])
}

Step 2: Modify Zustand Store Default Value

// stores/authStore.ts
export const useAuthStore = create<AuthState>()(
persist(
(set) => ({
user: null,
token: null,
loading: false, // 👈 Change to false, let useAuth hook control state
setUser: (user) => set({ user }),
setToken: (token) => set({ token }),
setLoading: (loading) => set({ loading }),
logout: () => set({ user: null, token: null, loading: false }),
}),
{
name: 'auth-storage',
partialize: (state) => ({ user: state.user, token: state.token }),
}
)
)

Step 3: Inject Test Auth in Playwright Fixture

// e2e/fixtures.ts
import { test as base } from '@playwright/test'

export const mockUser = {
id: 'test-user-id',
email: 'test@example.com',
created_at: '2024-01-01T00:00:00Z',
}

export const test = base.extend({
authenticatedPage: async ({ page }, use) => {
// Visit page first to set localStorage origin
await page.goto('/login')

// 👇 Inject test auth state into localStorage
await page.evaluate(
({ user, token }) => {
localStorage.setItem('test-auth-user', JSON.stringify(user))
localStorage.setItem('test-auth-token', token)
},
{ user: mockUser, token: 'mock-access-token' }
)

// Navigate to protected page, useAuth will detect test mode
await page.goto('/dashboard')

await use(page)
},
})

Step 4: Use in Tests

// e2e/dashboard.spec.ts
import { test, expect } from './fixtures'

test('dashboard shows user agents', async ({ authenticatedPage }) => {
// authenticatedPage is already authenticated, no login needed
await expect(authenticatedPage.getByText('Test Agent')).toBeVisible()
})

Complete Code Structure

agent-frontend/
├── e2e/
│ ├── fixtures.ts # Playwright fixture + mock data
│ ├── dashboard.spec.ts # Test cases
│ └── ...
├── src/
│ ├── hooks/
│ │ └── useAuth.ts # Test mode detection
│ └── stores/
│ └── authStore.ts # loading: false default
└── playwright.config.ts

Key Takeaways

  1. Use special prefix for test mode keys: test-auth-* won't appear in production
  2. Check before initialize: Check localStorage first, then proceed with Supabase Auth
  3. Skip auth listener: No need to listen for auth state changes in test mode
  4. Change loading default to false: Let the hook explicitly control loading state

Interested in similar solutions? Contact us

Implementing Data Caching in Zustand Store

· 3 min read

Encountered this issue while building an AI Agent platform. Here's the root cause and solution.

TL;DR

Add lastFetchTime field and TTL constant to Zustand store. Check cache expiration before requesting. Implement effective data caching in ~10 lines of code, avoiding duplicate requests across pages.

Problem

MCP tools list is used by multiple pages (Agent Settings, Tools Marketplace, Chat tool selector). Each page entry triggers an API request:

GET /api/mcp-tools  → 200  (AgentSettingsPage)
GET /api/mcp-tools → 200 (McpToolsPage)
GET /api/mcp-tools → 200 (ChatPage tool selector)

Tools list rarely changes (admin configures manually), but frequent requests waste bandwidth and slow page loads.

Root Cause

Components call API directly without caching:

// ❌ No caching: requests on every mount
function McpToolsPage() {
const [tools, setTools] = useState([])

useEffect(() => {
mcpToolsApi.list().then(setTools)
}, [])

return <ToolList tools={tools} />
}

Problems:

  1. Each page requests independently — No global state sharing
  2. Repeated requests in short time — Triggered when user navigates between pages
  3. No refresh control — Re-fetches even when data unchanged

Solution

Zustand Store + TTL Cache

// src/stores/mcpToolsStore.ts
import { create } from 'zustand'
import { mcpToolsApi, type McpTool } from '@/services/api'

const CACHE_TTL = 10 * 60 * 1000 // 10 minutes

interface McpToolsState {
tools: McpTool[]
lastFetchTime: number | null
loading: boolean
error: string | null

fetchTools: (force?: boolean) => Promise<void>
clearError: () => void
}

export const useMcpToolsStore = create<McpToolsState>((set, get) => ({
tools: [],
lastFetchTime: null,
loading: false,
error: null,

fetchTools: async (force = false) => {
const { tools, lastFetchTime } = get()

// Has cache, not expired, not forced → skip
if (tools.length && lastFetchTime && !force) {
if (Date.now() - lastFetchTime < CACHE_TTL) {
return // Cache hit
}
}

set({ loading: true, error: null })
try {
const data = await mcpToolsApi.list()
set({ tools: data, lastFetchTime: Date.now(), loading: false })
} catch (err) {
const message = err instanceof Error ? err.message : 'Failed to fetch tools'
set({ error: message, loading: false })
}
},

clearError: () => set({ error: null }),
}))

Component Usage

// ✅ Using store cache
function McpToolsPage() {
const { tools, loading, fetchTools } = useMcpToolsStore()

useEffect(() => {
fetchTools() // Auto-checks cache
}, [fetchTools])

if (loading) return <Spinner />
return <ToolList tools={tools} />
}

// Force refresh
function RefreshButton() {
const { fetchTools } = useMcpToolsStore()
return <button onClick={() => fetchTools(true)}>Refresh</button>
}

Core Logic Explained

// Cache check logic
if (tools.length && lastFetchTime && !force) {
if (Date.now() - lastFetchTime < CACHE_TTL) {
return // Cache valid, skip request
}
}
ConditionMeaning
tools.lengthHas data (empty array not valid cache)
lastFetchTimeRecorded last request time
!forceNot forced refresh
Date.now() - lastFetchTime < CACHE_TTLNot expired

Use Cases

ScenarioSuitableReason
Tool lists, config dictionaries✅ YesLow change frequency, shared across pages
User permissions, roles✅ YesRarely changes within session
Real-time data (messages, notifications)❌ NoNeeds latest state
Paginated lists❌ NoLarge data volume, complex caching strategy

Extension: Fine-grained Cache Control

interface CacheOptions {
ttl: number // Expiration time
staleWhileRevalidate: boolean // Return stale data while refreshing
}

// Background refresh after expiration, return cached data first
if (tools.length && lastFetchTime) {
const age = Date.now() - lastFetchTime
if (age < CACHE_TTL) {
return // Cache fresh
}
if (options.staleWhileRevalidate && age < CACHE_TTL * 2) {
// Cache stale but acceptable, background refresh
mcpToolsApi.list().then(data => set({ tools: data, lastFetchTime: Date.now() }))
return
}
}

Key Principles

  1. Set reasonable TTL — Based on data change frequency
  2. Provide force refresh — Users can manually get latest data
  3. Show loading on first fetch — Empty data shouldn't skip request

Interested in similar solutions? Contact us

Fix the Hidden Pitfall of httpx async with client.post()

· 2 min read

Encountered this issue while building a multi-service SaaS system. Documenting the root cause and solution.

TL;DR

Don't use async with client.post() pattern with httpx.AsyncClient. Create the client first, then call methods: response = await client.post().

Problem Symptoms

import httpx

async def call_api():
async with httpx.AsyncClient() as client:
async with client.post(url, json=data) as response: # Problem code
return response.json()

This code sometimes works, sometimes errors:

httpx.RemoteProtocolError: cannot write to closing transport
RuntimeError: Session is closed

Root Cause

The async with client.post() Trap

client.post() returns a Response object, not a context manager. Wrapping it with async with causes:

  1. Premature connection closure: The connection closes immediately when the async with block ends, but the response may still be reading
  2. Resource contention: With concurrent requests, connection pool state becomes chaotic

Understanding httpx Context Managers Correctly

# ✅ Correct: client is the context manager
async with httpx.AsyncClient() as client:
response = await client.post(url, json=data)
return response.json()

# ❌ Wrong: treating response as context manager
async with client.post(url) as response:
...

Solution

Option 1: Single Request (Simple Scenarios)

async def call_api(url: str, data: dict) -> dict:
async with httpx.AsyncClient() as client:
response = await client.post(url, json=data)
response.raise_for_status()
return response.json()

Option 2: Reuse Client (High-Frequency Requests)

# Global or dependency injection
_client = httpx.AsyncClient(timeout=30.0)

async def call_api(url: str, data: dict) -> dict:
response = await _client.post(url, json=data)
response.raise_for_status()
return response.json()

# On app shutdown
async def shutdown():
await _client.aclose()

Option 3: FastAPI Dependency Injection

from fastapi import Depends
from httpx import AsyncClient

async def get_http_client() -> AsyncClient:
async with AsyncClient(timeout=30.0) as client:
yield client

@router.post("/proxy")
async def proxy(
data: dict,
client: AsyncClient = Depends(get_http_client)
):
response = await client.post("https://external.api/endpoint", json=data)
return response.json()

FAQ

Q: How should httpx async with be used correctly?

A: async with is only for managing AsyncClient lifecycle, not wrapping individual requests. Correct pattern: async with AsyncClient() as client: response = await client.post(...).

Q: Why does async with client.post() sometimes work?

A: It may work by chance in single-threaded, low-concurrency scenarios, but will fail under high concurrency or network latency. This is a hidden bug—don't rely on it.

Q: How to configure httpx timeout?

A: AsyncClient(timeout=30.0) or AsyncClient(timeout=httpx.Timeout(connect=5.0, read=30.0)).

集成 Supabase Auth 到 FastAPI 的三个坑

· 4 min read

在为客户构建 SaaS 认证系统时遇到此问题,记录根因与解法。

TL;DR

Supabase Auth + FastAPI 集成有三个常见坑:JWKS 路径不是标准路径、ES256 签名需转换为 DER 格式、用户首次登录时本地数据库无记录。本文提供完整解决方案。

问题现象

坑 1:JWKS 路径 404

GET https://xxx.supabase.co/.well-known/jwks.json
# 404 Not Found

所有 JWT 验证请求返回 401 Invalid Token。

坑 2:ES256 签名验证失败

from jose import jwt
payload = jwt.decode(token, key, algorithms=["ES256"])
# JWTError: Signature verification failed

明明公钥是对的,但签名验证总是失败。

坑 3:用户首次登录无本地记录

# 创建 Agent 时
agent = Agent(user_id=current_user["user_id"], ...)
db.add(agent)
# ForeignKeyViolation: user_id 不存在

Supabase Auth 用户通过了 JWT 验证,但本地 agent_users 表没有该用户记录。

根因

坑 1:Supabase 非标准 JWKS 路径

标准 OAuth/OIDC 服务器 JWKS 在 /.well-known/jwks.json,但 Supabase 把认证服务放在 /auth/v1/ 子路径下:

标准路径Supabase 路径
/.well-known/jwks.json/auth/v1/.well-known/jwks.json

坑 2:ES256 原始签名 vs DER 格式

Supabase JWT 使用 ES256(P-256 曲线)签名。JWT 中的签名是 raw 格式r || s 拼接,64 字节),但 Python cryptography 库的 verify() 方法需要 DER-encoded ASN.1 格式

Raw:     r (32 bytes) || s (32 bytes) = 64 bytes
DER: 0x30 <len> 0x02 <r_len> <r> 0x02 <s_len> <s>

python-josejwt.decode() 在处理 ES256 时有兼容性问题,需要手动验证签名。

坑 3:认证与数据分离

Supabase Auth 是独立服务,用户注册/登录后只存在于 Supabase 的 auth.users 表。本地数据库的 agent_users 表需要手动同步。

解决方案

1. 正确的 JWKS URL

# config.py
class Settings(BaseSettings):
supabase_url: str = "https://xxx.supabase.co"

@property
def jwks_url(self) -> str:
# 关键:/auth/v1/ 前缀
return f"{self.supabase_url}/auth/v1/.well-known/jwks.json"

2. ES256 签名验证(完整代码)

import json
import base64
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric.utils import encode_dss_signature

def _base64url_decode(data: str) -> bytes:
"""Base64url 解码,自动补 padding"""
rem = len(data) % 4
if rem > 0:
data += "=" * (4 - rem)
return base64.urlsafe_b64decode(data)

def _raw_to_der_signature(raw_sig: bytes) -> bytes:
"""将 raw ECDSA 签名 (r||s) 转为 DER 格式"""
# P-256: r 和 s 各 32 字节
r = int.from_bytes(raw_sig[:32], "big")
s = int.from_bytes(raw_sig[32:], "big")
return encode_dss_signature(r, s)

def verify_es256_signature(token: str, public_key_jwk: dict) -> dict:
"""验证 ES256 JWT 签名,返回 payload"""
parts = token.split(".")
if len(parts) != 3:
raise ValueError("Invalid JWT format")

header_b64, payload_b64, signature_b64 = parts

# 1. 构建 EC 公钥
x = _base64url_decode(public_key_jwk["x"])
y = _base64url_decode(public_key_jwk["y"])
x_int = int.from_bytes(x, "big")
y_int = int.from_bytes(y, "big")

public_key = ec.EllipticCurvePublicNumbers(
x_int, y_int, ec.SECP256R1()
).public_key(default_backend())

# 2. 验证签名
message = f"{header_b64}.{payload_b64}".encode()
raw_signature = _base64url_decode(signature_b64)
der_signature = _raw_to_der_signature(raw_signature)

public_key.verify(
der_signature,
message,
ec.ECDSA(hashes.SHA256())
)

# 3. 返回 payload
return json.loads(_base64url_decode(payload_b64))

3. 用户同步服务

# app/services/user_service.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.user import AgentUser

async def ensure_user_exists(
db: AsyncSession,
user_id: str,
email: str,
plan: str = "free"
) -> AgentUser:
"""确保用户存在于本地数据库(从 Supabase Auth 同步)"""
# 检查是否存在
result = await db.execute(
select(AgentUser).where(AgentUser.user_id == user_id)
)
user = result.scalar_one_or_none()

if user:
return user

# 创建新用户
user = AgentUser(
user_id=user_id,
email=email,
plan=plan,
role="user"
)
db.add(user)
await db.commit()
await db.refresh(user)
return user

4. 在创建资源前调用

# app/routers/agents.py
@router.post("/")
async def create_agent(
input: CreateAgentInput,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user)
):
# 关键:确保用户存在
user = await ensure_user_exists(
db,
user_id=current_user["user_id"],
email=current_user["email"],
plan=current_user["plan"]
)

# 现在可以安全创建 Agent
agent = Agent(
user_id=user.user_id,
name=input.name,
llm_config=input.llm_config.model_dump()
)
...

FAQ

Q: Supabase JWT 验证返回 404 怎么办?

A: Supabase 的 JWKS 路径是 /auth/v1/.well-known/jwks.json,不是标准的 /.well-known/jwks.json。检查你的 JWKS URL 配置。

Q: python-jose 验证 ES256 签名失败怎么解决?

A: python-jose 对 ES256 支持不完善。使用 cryptography 库手动验证,需要将 JWT 的 raw 签名(r||s 64字节)转换为 DER 格式。

Q: FastAPI 如何同步 Supabase Auth 用户到本地数据库?

A: 在需要用户记录的 API(如创建资源)入口处调用 ensure_user_exists(),从 JWT 提取用户信息并同步到本地表。

Q: Supabase JWT 中的 user_id 在哪个字段?

A: sub 字段包含用户 UUID,email 字段包含邮箱,app_metadata.plan 包含订阅计划(自定义字段)。