mcpField
T
tRPC

tRPC

github.com/oriolrius开发2447 次安装2 周前更新
代码仓库

关于 tRPC

面向 tRPC 的专业指导技能，介绍如何构建端到端类型安全的 TypeScript API，无需代码生成。内容涵盖服务端与客户端设置、路由、验证、中间件和测试。

主要功能

  1. 01端到端 TypeScript 类型推断
  2. 02路由器与过程设计
  3. 03使用 Zod 验证输入
  4. 04中间件与访问控制
  5. 05客户端查询与变更
  6. 06过程测试

适用场景

  1. 01构建类型安全的 TypeScript API
  2. 02用嵌套路由组织 API
  3. 03验证输入并保护过程
  4. 04在 React 组件中调用 API

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/oriolrius/pki-manager-web --skill trpc

安装后在 Claude Code 中输入/trpc确认已加载。

详细介绍

tRPC

tRPC 专家辅助——使用 TypeScript 构建端到端类型安全的 API。

概述

tRPC 支持构建完全类型安全的 API，无需架构或代码生成：

  • 从服务端到客户端的完整 TypeScript 类型推断
  • 无需代码生成
  • 出色的开发体验，具备自动补全和类型安全
  • 与 Next.js、React Query 等配合出色

快速入门

安装
# Core packages
npm install @trpc/server@next @trpc/client@next @trpc/react-query@next

# Peer dependencies
npm install @tanstack/react-query@latest zod
基本设置（Next.js App Router）

1. 创建 tRPC Router

// server/trpc.ts
import { initTRPC } from '@trpc/server'
import { z } from 'zod'

const t = initTRPC.create()

export const router = t.router
export const publicProcedure = t.procedure

2. 定义 API Router

// server/routers/_app.ts
import { router, publicProcedure } from '../trpc'
import { z } from 'zod'

export const appRouter = router({
  hello: publicProcedure
    .input(z.object({ name: z.string() }))
    .query(({ input }) => {
      return { greeting: `Hello ${input.name}!` }
    }),

  createUser: publicProcedure
    .input(z.object({
      name: z.string(),
      email: z.string().email(),
    }))
    .mutation(async ({ input }) => {
      const user = await db.user.create({ data: input })
      return user
    }),
})

export type AppRouter = typeof appRouter

3. 创建 API Route

// app/api/trpc/[trpc]/route.ts
import { fetchRequestHandler } from '@trpc/server/adapters/fetch'
import { appRouter } from '@/server/routers/_app'

const handler = (req: Request) =>
  fetchRequestHandler({
    endpoint: '/api/trpc',
    req,
    router: appRouter,
    createContext: () => ({}),
  })

export { handler as GET, handler as POST }

4. 设置客户端 Provider

// app/providers.tsx
'use client'

import { QueryClient, QueryClientProvider } from '@tanstack/react-query'
import { httpBatchLink } from '@trpc/client'
import { useState } from 'react'
import { trpc } from '@/lib/trpc'

export function Providers({ children }: { children: React.ReactNode }) {
  const [queryClient] = useState(() => new QueryClient())
  const [trpcClient] = useState(() =>
    trpc.createClient({
      links: [
        httpBatchLink({
          url: 'http://localhost:3000/api/trpc',
        }),
      ],
    })
  )

  return (
    <trpc.Provider client={trpcClient} queryClient={queryClient}>
      <QueryClientProvider client={queryClient}>
        {children}
      </QueryClientProvider>
    </trpc.Provider>
  )
}

5. 创建 tRPC 客户端

// lib/trpc.ts
import { createTRPCReact } from '@trpc/react-query'
import type { AppRouter } from '@/server/routers/_app'

export const trpc = createTRPCReact<AppRouter>()

6. 在组件中使用

'use client'

import { trpc } from '@/lib/trpc'

export default function Home() {
  const hello = trpc.hello.useQuery({ name: 'World' })
  const createUser = trpc.createUser.useMutation()

  return (
    <div>
      <p>{hello.data?.greeting}</p>
      <button
        onClick={() => createUser.mutate({
          name: 'John',
          email: '[email protected]'
        })}
      >
        Create User
      </button>
    </div>
  )
}

Router 定义

基本 Router
import { router, publicProcedure } from './trpc'
import { z } from 'zod'

export const userRouter = router({
  // Query - for fetching data
  getById: publicProcedure
    .input(z.string())
    .query(async ({ input }) => {
      return await db.user.findUnique({ where: { id: input } })
    }),

  // Mutation - for creating/updating/deleting
  create: publicProcedure
    .input(z.object({
      name: z.string(),
      email: z.string().email(),
    }))
    .mutation(async ({ input }) => {
      return await db.user.create({ data: input })
    }),

  // Subscription - for real-time updates
  onUpdate: publicProcedure
    .subscription(() => {
      return observable<User>((emit) => {
        // Implementation
      })
    }),
})
嵌套 Router
import { router } from './trpc'
import { userRouter } from './routers/user'
import { postRouter } from './routers/post'
import { commentRouter } from './routers/comment'

export const appRouter = router({
  user: userRouter,
  post: postRouter,
  comment: commentRouter,
})

// Usage on client:
// trpc.user.getById.useQuery('123')
// trpc.post.list.useQuery()
// trpc.comment.create.useMutation()
合并 Router
import { router, publicProcedure } from './trpc'

const userRouter = router({
  list: publicProcedure.query(() => {/* ... */}),
  getById: publicProcedure.input(z.string()).query(() => {/* ... */}),
})

const postRouter = router({
  list: publicProcedure.query(() => {/* ... */}),
  create: publicProcedure.input(z.object({})).mutation(() => {/* ... */}),
})

// Merge into app router
export const appRouter = router({
  user: userRouter,
  post: postRouter,
})

使用 Zod 验证输入

基本验证
import { z } from 'zod'

export const userRouter = router({
  create: publicProcedure
    .input(z.object({
      name: z.string().min(2).max(50),
      email: z.string().email(),
      age: z.number().int().positive().optional(),
      role: z.enum(['user', 'admin']),
    }))
    .mutation(async ({ input }) => {
      // input is fully typed!
      return await db.user.create({ data: input })
    }),
})
复杂验证
const createPostInput = z.object({
  title: z.string().min(5).max(100),
  content: z.string().min(10),
  published: z.boolean().default(false),
  tags: z.array(z.string()).min(1).max(5),
  metadata: z.object({
    views: z.number().default(0),
    likes: z.number().default(0),
  }).optional(),
})

export const postRouter = router({
  create: publicProcedure
    .input(createPostInput)
    .mutation(async ({ input }) => {
      return await db.post.create({ data: input })
    }),
})
可复用架构
// schemas/user.ts
export const userSchema = z.object({
  id: z.string(),
  name: z.string(),
  email: z.string().email(),
})

export const createUserSchema = userSchema.omit({ id: true })
export const updateUserSchema = userSchema.partial()

// Use in router
export const userRouter = router({
  create: publicProcedure
    .input(createUserSchema)
    .mutation(({ input }) => {/* ... */}),

  update: publicProcedure
    .input(z.object({
      id: z.string(),
      data: updateUserSchema,
    }))
    .mutation(({ input }) => {/* ... */}),
})

上下文

创建上下文
// server/context.ts
import { inferAsyncReturnType } from '@trpc/server'
import { FetchCreateContextFnOptions } from '@trpc/server/adapters/fetch'

export async function createContext(opts: FetchCreateContextFnOptions) {
  // Get session from cookies/headers
  const session = await getSession(opts.req)

  return {
    session,
    db,
  }
}

export type Context = inferAsyncReturnType<typeof createContext>
在 tRPC 中使用上下文
// server/trpc.ts
import { initTRPC } from '@trpc/server'
import { Context } from './context'

const t = initTRPC.context<Context>().create()

export const router = t.router
export const publicProcedure = t.procedure
在 Procedures 中访问上下文
export const userRouter = router({
  me: publicProcedure.query(({ ctx }) => {
    // ctx.session, ctx.db are available
    if (!ctx.session) {
      throw new TRPCError({ code: 'UNAUTHORIZED' })
    }

    return ctx.db.user.findUnique({
      where: { id: ctx.session.userId }
    })
  }),
})

中间件

创建中间件
// server/trpc.ts
import { initTRPC, TRPCError } from '@trpc/server'

const t = initTRPC.context<Context>().create()

// Logging middleware
const loggerMiddleware = t.middleware(async ({ path, type, next }) => {
  const start = Date.now()
  const result = await next()
  const duration = Date.now() - start

  console.log(`${type} ${path} took ${duration}ms`)

  return result
})

// Auth middleware
const isAuthed = t.middleware(({ ctx, next }) => {
  if (!ctx.session) {
    throw new TRPCError({ code: 'UNAUTHORIZED' })
  }

  return next({
    ctx: {
      // Infers session is non-nullable
      session: ctx.session,
    },
  })
})

// Create procedures with middleware
export const publicProcedure = t.procedure.use(loggerMiddleware)
export const protectedProcedure = t.procedure.use(loggerMiddleware).use(isAuthed)
使用受保护的 Procedures
export const postRouter = router({
  // Public - anyone can access
  list: publicProcedure.query(() => {
    return db.post.findMany({ where: { published: true } })
  }),

  // Protected - requires authentication
  create: protectedProcedure
    .input(z.object({ title: z.string() }))
    .mutation(({ ctx, input }) => {
      // ctx.session is guaranteed to exist
      return db.post.create({
        data: {
          ...input,
          authorId: ctx.session.userId,
        },
      })
    }),
})
基于角色的中间件
const requireRole = (role: string) =>
  t.middleware(({ ctx, next }) => {
    if (!ctx.session || ctx.session.role !== role) {
      throw new TRPCError({ code: 'FORBIDDEN' })
    }
    return next()
  })

export const adminProcedure = protectedProcedure.use(requireRole('admin'))

export const userRouter = router({
  delete: adminProcedure
    .input(z.string())
    .mutation(({ input }) => {
      return db.user.delete({ where: { id: input } })
    }),
})

客户端用法

查询
'use client'

import { trpc } from '@/lib/trpc'

export default function UserList() {
  // Basic query
  const users = trpc.user.list.useQuery()

  // Query with input
  const user = trpc.user.getById.useQuery('user-123')

  // Disabled query
  const profile = trpc.user.getProfile.useQuery(
    { id: userId },
    { enabled: !!userId }
  )

  // With options
  const posts = trpc.post.list.useQuery(undefined, {
    refetchInterval: 5000,
    staleTime: 1000,
  })

  if (users.isLoading) return <div>Loading...</div>
  if (users.error) return <div>Error: {users.error.message}</div>

  return (
    <ul>
      {users.data?.map(user => (
        <li key={user.id}>{user.name}</li>
      ))}
    </ul>
  )
}
修改
'use client'

export default function CreateUser() {
  const utils = trpc.useContext()

  const createUser = trpc.user.create.useMutation({
    onSuccess: () => {
      // Invalidate and refetch
      utils.user.list.invalidate()
    },
    onError: (error) => {
      console.error('Failed to create user:', error)
    },
  })

  const handleSubmit = (e: React.FormEvent<HTMLFormElement>) => {
    e.preventDefault()
    const formData = new FormData(e.currentTarget)

    createUser.mutate({
      name: formData.get('name') as string,
      email: formData.get('email') as string,
    })
  }

  return (
    <form onSubmit={handleSubmit}>
      <input name="name" required />
      <input name="email" type="email" required />
      <button type="submit" disabled={createUser.isLoading}>
        {createUser.isLoading ? 'Creating...' : 'Create User'}
      </button>
    </form>
  )
}
乐观更新
const updatePost = trpc.post.update.useMutation({
  onMutate: async (newPost) => {
    // Cancel outgoing refetches
    await utils.post.list.cancel()

    // Snapshot previous value
    const previousPosts = utils.post.list.getData()

    // Optimistically update
    utils.post.list.setData(undefined, (old) =>
      old?.map(post =>
        post.id === newPost.id ? { ...post, ...newPost } : post
      )
    )

    return { previousPosts }
  },
  onError: (err, newPost, context) => {
    // Rollback on error
    utils.post.list.setData(undefined, context?.previousPosts)
  },
  onSettled: () => {
    // Refetch after success or error
    utils.post.list.invalidate()
  },
})
无限查询
// Server
export const postRouter = router({
  list: publicProcedure
    .input(z.object({
      cursor: z.string().optional(),
      limit: z.number().min(1).max(100).default(10),
    }))
    .query(async ({ input }) => {
      const posts = await db.post.findMany({
        take: input.limit + 1,
        cursor: input.cursor ? { id: input.cursor } : undefined,
      })

      let nextCursor: string | undefined = undefined
      if (posts.length > input.limit) {
        const nextItem = posts.pop()
        nextCursor = nextItem!.id
      }

      return { posts, nextCursor }
    }),
})

// Client
export default function InfinitePosts() {
  const posts = trpc.post.list.useInfiniteQuery(
    { limit: 10 },
    {
      getNextPageParam: (lastPage) => lastPage.nextCursor,
    }
  )

  return (
    <div>
      {posts.data?.pages.map((page, i) => (
        <div key={i}>
          {page.posts.map(post => (
            <div key={post.id}>{post.title}</div>
          ))}
        </div>
      ))}

      <button
        onClick={() => posts.fetchNextPage()}
        disabled={!posts.hasNextPage || posts.isFetchingNextPage}
      >
        {posts.isFetchingNextPage ? 'Loading...' : 'Load More'}
      </button>
    </div>
  )
}

错误处理

服务端错误
import { TRPCError } from '@trpc/server'

export const postRouter = router({
  getById: publicProcedure
    .input(z.string())
    .query(async ({ input }) => {
      const post = await db.post.findUnique({ where: { id: input } })

      if (!post) {
        throw new TRPCError({
          code: 'NOT_FOUND',
          message: 'Post not found',
        })
      }

      return post
    }),

  create: protectedProcedure
    .input(z.object({ title: z.string() }))
    .mutation(async ({ ctx, input }) => {
      if (!ctx.session.verified) {
        throw new TRPCError({
          code: 'FORBIDDEN',
          message: 'Email must be verified',
        })
      }

      try {
        return await db.post.create({ data: input })
      } catch (error) {
        throw new TRPCError({
          code: 'INTERNAL_SERVER_ERROR',
          message: 'Failed to create post',
          cause: error,
        })
      }
    }),
})
错误代码
  • BAD_REQUEST - 输入无效
  • UNAUTHORIZED - 未通过身份验证
  • FORBIDDEN - 未获授权
  • NOT_FOUND - 未找到资源
  • TIMEOUT - 请求超时
  • CONFLICT - 资源冲突
  • PRECONDITION_FAILED - 前置条件检查失败
  • PAYLOAD_TOO_LARGE - 请求过大
  • METHOD_NOT_SUPPORTED - 不支持此 HTTP 方法
  • TOO_MANY_REQUESTS - 请求频率受限
  • CLIENT_CLOSED_REQUEST - 客户端已关闭请求
  • INTERNAL_SERVER_ERROR - 服务端错误
客户端错误处理
const createPost = trpc.post.create.useMutation({
  onError: (error) => {
    if (error.data?.code === 'UNAUTHORIZED') {
      router.push('/login')
    } else if (error.data?.code === 'FORBIDDEN') {
      alert('You do not have permission')
    } else {
      alert('Something went wrong')
    }
  },
})

服务端调用

在服务端组件中
// app/users/page.tsx
import { createCaller } from '@/server/routers/_app'
import { createContext } from '@/server/context'

export default async function UsersPage() {
  const ctx = await createContext({ req: {} as any })
  const caller = createCaller(ctx)

  const users = await caller.user.list()

  return (
    <ul>
      {users.map(user => (
        <li key={user.id}>{user.name}</li>
      ))}
    </ul>
  )
}
创建 Caller
// server/routers/_app.ts
export const createCaller = createCallerFactory(appRouter)

// Usage
const caller = createCaller(ctx)
const user = await caller.user.getById('123')

高级模式

请求批处理
import { httpBatchLink } from '@trpc/client'

const trpcClient = trpc.createClient({
  links: [
    httpBatchLink({
      url: '/api/trpc',
      maxURLLength: 2083, // Reasonable limit
    }),
  ],
})
请求去重

由 React Query 自动处理——多个组件请求相同数据时只会发出一个请求。

自定义标头
const trpcClient = trpc.createClient({
  links: [
    httpBatchLink({
      url: '/api/trpc',
      headers: () => {
        return {
          Authorization: `Bearer ${getToken()}`,
        }
      },
    }),
  ],
})
错误格式化
// server/trpc.ts
const t = initTRPC.context<Context>().create({
  errorFormatter({ shape, error }) {
    return {
      ...shape,
      data: {
        ...shape.data,
        zodError:
          error.cause instanceof ZodError
            ? error.cause.flatten()
            : null,
      },
    }
  },
})

测试

测试 Procedures
import { appRouter } from '@/server/routers/_app'
import { createCaller } from '@/server/routers/_app'

describe('user router', () => {
  it('creates user', async () => {
    const ctx = { session: mockSession, db: mockDb }
    const caller = createCaller(ctx)

    const user = await caller.user.create({
      name: 'John',
      email: '[email protected]',
    })

    expect(user.name).toBe('John')
  })
})

最佳实践

  1. 使用 Zod 进行验证 - 始终验证输入
  2. 保持 Procedures 简短 - 遵循单一职责原则
  3. 使用中间件进行身份验证 - 不要重复身份验证检查
  4. 为上下文添加类型 - 实现完整的类型安全
  5. 组织 Router - 按逻辑领域拆分
  6. 妥善处理错误 - 使用合适的错误代码
  7. 充分利用 React Query - 使用其缓存和重新获取功能
  8. 批处理请求 - 启用批处理以提升性能
  9. 使用乐观更新 - 改善用户体验
  10. 为 Procedures 编写文档 - 添加 JSDoc 注释

资源

评论

支持 Markdown