Skip to main content

9 posts tagged with "saas-development"

View all tags

Save 70% of Your AI Coding Plan Quota by Replacing ZhiPu Search with Tavily MCP

· 3 min read

When using ZhiPu GLM Coding Plan with Claude Code, I noticed my monthly quota draining faster than expected. The culprit? Built-in MCP tools — web-search-prime and web-reader — were consuming coding conversation quota for every web search and page fetch.

TL;DR

Replace ZhiPu's built-in web-search-prime and web-reader MCP services with Tavily MCP. Result: free 1000 searches/month with zero impact on coding quota.

The Problem: Every Search Costs Quota

ZhiPu GLM Coding Plan includes several built-in MCP services:

web-search-prime   → web search (consumes quota)
web-reader → web page reading (consumes quota)
zread → GitHub repo reading (consumes quota)

In my case, searching WooCommerce documentation, reading API pages, and checking marketplace standards during theme development consumed roughly 30% of my monthly MCP quota in just a few days.

MCP usage over one week showing search and reader consuming most calls

The 16% / 30% usage shown above was reached in less than a week, with web search being the primary consumer.

Solution: Tavily MCP

Tavily provides a free tier with 1000 API calls/month, completely independent of your coding plan quota.

Step 1: Get Tavily API Key

Register at tavily.com, free tier includes 1000 calls/month.

Step 2: Add Tavily MCP to Claude Code

claude mcp add tavily-search -s user -- npx -y tavily-mcp@latest

Step 3: Configure API Key

Edit ~/.claude.json, find the tavily-search entry and add the environment variable:

{
"mcpServers": {
"tavily-search": {
"type": "stdio",
"command": "npx",
"args": ["-y", "tavily-mcp@latest"],
"env": {
"TAVILY_API_KEY": "tvly-your-key-here"
}
}
}
}

Step 4: Remove ZhiPu's search services

claude mcp remove web-search-prime -s user
claude mcp remove web-reader -s user

Also clean up settings.json permissions — remove mcp__web-search-prime__* and mcp__web-reader__* from the allow list, add mcp__tavily-search__* instead.

Step 5: Verify

claude mcp list
# Should show: tavily-search: npx -y tavily-mcp@latest - Connected
# Should NOT show: web-search-prime, web-reader

Comparison

FeatureZhiPu Built-inTavily MCP
CostConsumes coding plan quotaFree 1000 calls/month
Search qualityOptimized for ChineseBalanced for EN/CN
Page extractionSeparate toolBuilt-in extract
ConfigurationPlatform-injected, can't disableOne claude mcp add command
After free tierSqueezes coding conversation budgetPay-as-you-go continues

Notes

Important Notes

  1. ZhiPu services are platform-injected — they may reappear after restart even after removal. Add them to settings.json deny list to block permanently.
  2. Keep zread (GitHub repo reading) — low consumption and unique functionality.
  3. Keep doubao-vision (image analysis) — Tavily doesn't cover this scenario.

Recommendation

Also using ZhiPu GLM Coding Plan?

Learn More About ZhiPu GLM Coding Plan


Build a Custom MCP Toolkit with Python FastMCP to Connect Any AI Model

· 8 min read

While building AI Agent systems for clients, we found that different tasks require vastly different model capabilities and costs: vision models for image analysis, lightweight models for text completion, and local models for internal data queries. MCP (Model Context Protocol) turns each capability into an independent tool that AI clients invoke on demand.

TL;DR

Build a custom MCP Server in 30 minutes with Python FastMCP, connecting any OpenAI-compatible API based on scenario and cost. This article demonstrates the full workflow using the Doubao vision model, with extension templates for text generation, image generation, TTS, and more.

What Problem Does MCP Solve?

MCP (Model Context Protocol) is an open protocol proposed by Anthropic that standardizes communication between AI applications and external tools. Think of it as USB-C: regardless of the device, plug it in and it works.

Traditionally, integrating each AI capability requires writing custom integration code. MCP standardizes this process:

AI Client (Claude Code / Cursor / Cherry Studio)
↓ MCP Protocol (stdio / SSE)
MCP Server (your toolkit)
↓ HTTP API
Various AI Models / Internal APIs

One MCP Server can expose multiple tools, each backed by a different model or API.

Why Python over Node.js?

The MCP official SDK supports both Python and TypeScript. Both are functionally equivalent, so the choice depends on your context:

DimensionPython (FastMCP)Node.js (@modelcontextprotocol/sdk)
AI Ecosystemhttpx/OpenAI SDK native supportAdditional dependencies needed
Code VolumeOne-line decorator to define a toolManual schema registration required
Binary Handlingbase64/PIL is conciseBuffer API slightly more verbose
Data Sciencepandas/numpy readily availableNeeds separate toolchain
Deploymentvenv isolation, no Node version issuesNode version compatibility is a common pain point

Bottom line: MCP Server's job is "call APIs + process data." Python is more concise and intuitive for HTTP calls, image/file handling, and data transformations, with fewer dependencies. Node.js is a better fit for teams already invested in TypeScript.

30-Minute Build: Image Analysis Tool

Setup

mkdir -p ~/.claude/mcp-servers/doubao-vision
cd ~/.claude/mcp-servers/doubao-vision
python3 -m venv venv
source venv/bin/activate
pip install mcp httpx

Server Code

"""Doubao Vision MCP Server - OpenAI-compatible image analysis via Volcengine Ark."""

import base64
import os
from pathlib import Path

import httpx
from mcp.server.fastmcp import FastMCP

# Config - inject via environment variables, never hardcode
BASE_URL = os.getenv("DOUBAO_BASE_URL", "https://ark.cn-beijing.volces.com/api/v3")
API_KEY = os.getenv("DOUBAO_API_KEY", "")
MODEL = os.getenv("DOUBAO_MODEL", "doubao-1-5-vision-pro-32k-250115")

mcp = FastMCP("doubao-vision")


def _build_image_content(image_source: str) -> dict:
"""Build image content part from URL or local file path."""
if image_source.startswith(("http://", "https://")):
return {"type": "image_url", "image_url": {"url": image_source}}
# Local file: encode to base64 data URI
path = Path(image_source)
if not path.exists():
raise FileNotFoundError(f"Image not found: {image_source}")
mime_map = {
".jpg": "image/jpeg", ".jpeg": "image/jpeg",
".png": "image/png", ".gif": "image/gif",
".webp": "image/webp",
}
mime = mime_map.get(path.suffix.lower(), "image/png")
data = base64.b64encode(path.read_bytes()).decode()
return {
"type": "image_url",
"image_url": {"url": f"data:{mime};base64,{data}"},
}


@mcp.tool()
def analyze_image(image_source: str, prompt: str) -> str:
"""Analyze an image using Doubao Vision Pro model.

Supports remote URLs and local file paths.

Args:
image_source: URL or local file path to the image.
prompt: What to analyze or describe about the image.
"""
try:
image_content = _build_image_content(image_source)
except FileNotFoundError as e:
return f"Error: {e}"

messages = [
{
"role": "user",
"content": [
{"type": "text", "text": prompt},
image_content,
],
}
]

resp = httpx.post(
f"{BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
},
json={
"model": MODEL,
"messages": messages,
"max_tokens": 4096,
},
timeout=60,
)

if resp.status_code != 200:
return f"API error {resp.status_code}: {resp.text}"

data = resp.json()
return data["choices"][0]["message"]["content"]


if __name__ == "__main__":
mcp.run(transport="stdio")

Key design decisions:

  • _build_image_content: Unified handling for remote URLs and local files (auto base64 encoding)
  • @mcp.tool() decorator: Function signatures and docstrings auto-generate MCP tool descriptions; AI clients discover parameters automatically
  • Environment variable config: API keys are never hardcoded; injected via .mcp.json env field

Register with Client

Add to ~/.claude/.mcp.json (global) or project-level .mcp.json:

{
"mcpServers": {
"doubao-vision": {
"command": "/path/to/venv/bin/python",
"args": ["/path/to/server.py"],
"env": {
"DOUBAO_API_KEY": "your-api-key",
"DOUBAO_MODEL": "doubao-1-5-vision-pro-32k-250115",
"DOUBAO_BASE_URL": "https://ark.cn-beijing.volces.com/api/v3"
}
}
}
}

Extend on Demand: Multi-Scenario Toolkit

One MCP Server can register multiple tools, each using a different model. Here are templates for various scenarios:

Text Generation / Completion

@mcp.tool()
def generate_text(prompt: str, model: str = "deepseek-chat") -> str:
"""Generate text using specified model. Supports deepseek-chat, qwen-turbo, etc."""
providers = {
"deepseek-chat": {"base": "https://api.deepseek.com/v1", "key": os.getenv("DEEPSEEK_API_KEY")},
"qwen-turbo": {"base": "https://dashscope.aliyuncs.com/compatible-mode/v1", "key": os.getenv("QWEN_API_KEY")},
}
cfg = providers.get(model)
if not cfg:
return f"Unknown model: {model}"
# Call OpenAI-compatible endpoint...

Image Generation

@mcp.tool()
def generate_image(prompt: str, size: str = "1024x1024") -> str:
"""Generate image from text prompt using Stable Diffusion WebUI API."""
resp = httpx.post(
"http://localhost:7860/sdapi/v1/txt2img",
json={"prompt": prompt, "width": 1024, "height": 1024},
timeout=120,
)
data = resp.json()
img_bytes = base64.b64decode(data["images"][0])
path = f"/tmp/mcp-gen-{hash(prompt)}.png"
Path(path).write_bytes(img_bytes)
return f"Image saved to: {path}"

Text-to-Speech (TTS)

@mcp.tool()
def text_to_speech(text: str, voice: str = "default") -> str:
"""Convert text to speech audio file."""
resp = httpx.post(
"https://openspeech.bytedance.com/api/v1/tts",
headers={"Authorization": f"Bearer;{os.getenv('TTS_API_KEY')}"},
json={"text": text, "voice": voice, "format": "mp3"},
timeout=30,
)
path = f"/tmp/mcp-tts-{hash(text)}.mp3"
Path(path).write_bytes(resp.content)
return f"Audio saved to: {path}"

Internal API Wrapper

@mcp.tool()
def query_database(sql: str) -> str:
"""Execute read-only SQL query on internal database. SELECT only."""
if not sql.strip().upper().startswith("SELECT"):
return "Error: Only SELECT queries are allowed."
resp = httpx.post(
"http://internal-api.company.com/query",
json={"sql": sql},
headers={"Authorization": f"Bearer {os.getenv('INTERNAL_API_KEY')}"},
timeout=10,
)
return resp.json()["result"]

Cost Optimization: Multi-Model Routing

Using different cost-level models for different tasks is the core value of a custom toolkit:

Task TypeRecommended ModelCost Level
Simple classification/extractionQwen-Turbo / Ollama localVery low
Copywriting/translationDeepSeek-ChatLow
Image analysisDoubao Vision ProMedium
Complex reasoningClaude / GPT-4High

Implement routing in your MCP Server:

@mcp.tool()
def smart_query(task: str, content: str) -> str:
"""Route task to optimal model based on complexity."""
route = {
"classify": ("qwen-turbo", "https://dashscope.aliyuncs.com/compatible-mode/v1"),
"analyze_image": ("doubao-1-5-vision-pro-32k-250115", "https://ark.cn-beijing.volces.com/api/v3"),
"deep_reason": ("deepseek-reasoner", "https://api.deepseek.com/v1"),
}
model, base = route.get(task, route["classify"])
# Unified OpenAI-compatible call...

Best Practices

  1. API Key Security: Inject via environment variables; never hardcode in source or commit to git
  2. Timeout Control: Set 120s+ for slow tasks (image generation), 10-30s for simple text tasks
  3. Error Handling: Return clear error messages; AI clients display them to users
  4. Sandbox Limits: Restrict scope when wrapping internal APIs (e.g., SQL SELECT only)
  5. Concurrency: Synchronous httpx calls are fine; MCP protocol itself is serial

Extensibility: Not Just Tools, but AI Infrastructure

The examples above cover a single MCP Server. In production, extensibility goes much further:

Multi-Server Composition

Split different capabilities into independent servers, load on demand:

{
"mcpServers": {
"doubao-vision": { "command": "python", "args": ["vision.py"] },
"deepseek-text": { "command": "python", "args": ["text.py"] },
"internal-api": { "command": "python", "args": ["api.py"] }
}
}

Benefit: one server crashing doesn't affect others; different team members maintain different servers.

Shared Configuration

Multiple servers reuse the same provider config via a shared Python module:

# providers.py - centralized API config management
PROVIDERS = {
"doubao": {
"base_url": "https://ark.cn-beijing.volces.com/api/v3",
"api_key_env": "DOUBAO_API_KEY",
},
"deepseek": {
"base_url": "https://api.deepseek.com/v1",
"api_key_env": "DEEPSEEK_API_KEY",
},
}

MCP Resources and Prompts

MCP isn't just about Tools. It also supports Resources (static data) and Prompts (preset templates):

# Resources: let AI read your internal docs
@mcp.resource("docs://api-spec")
def get_api_spec() -> str:
return Path("openapi.yaml").read_text()

# Prompts: preset common analysis templates
@mcp.prompt()
def code_review_prompt(code: str) -> str:
return f"Review this code for security issues and performance:\n\n{code}"

From Local to Remote Deployment

stdio transport works for local development. For production, switch to SSE transport and deploy as an HTTP service:

# Local development
mcp.run(transport="stdio")

# Production deployment (remote AI clients access via HTTP)
mcp.run(transport="sse", host="0.0.0.0", port=8080)

Advanced Directions

DirectionDescription
Caching LayerCache identical requests to reduce API costs
Rate LimitingLimit call frequency per user or tool
Audit LoggingLog every tool call's input/output for debugging and compliance
StreamingUse SSE streaming for long text generation to reduce wait time
Tool PipelinesChain one tool's output as another's input

Interested in similar solutions? Get in touch

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"]', '[email protected]')
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: '[email protected]',
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: '[email protected]',
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 包含订阅计划(自定义字段)。