tRPC 完全指南：TypeScript 端到端类型安全 API

TL;DR

tRPC 让你无需编写 API schema、代码生成或 REST/GraphQL 样板代码即可构建完全类型安全的 API。你在服务器上定义过程，然后从客户端调用，享有完整的自动补全和类型检查。API 的更改在编译时被 TypeScript 捕获，而不是在生产中由用户发现。它原生集成 Zod 进行验证，集成 React Query 进行数据获取。

Key Takeaways

tRPC 无需 schema 或代码生成即可提供从服务器到客户端的端到端类型安全
基于 React Query 构建，免费获得缓存、去重和后台重新获取
Zod 集成使输入验证类型安全且声明式
中间件和上下文实现认证、授权和请求范围数据
支持 Next.js App Router、Express、Fastify 等框架
通过 WebSocket 订阅实现实时功能

什么是 tRPC？

tRPC（TypeScript 远程过程调用）是一个库，让你为 TypeScript 应用构建类型安全的 API。与 REST 或 GraphQL 不同，tRPC 不需要 API schema 定义、代码生成步骤或运行时类型检查开销。服务器端过程定义成为唯一的真相来源。

使用 tRPC，当你在服务器上重命名字段时，TypeScript 会立即标记每个需要更新的客户端用法。当你添加必需的输入参数时，客户端代码在提供之前无法编译。

// Traditional REST workflow:
// 1. Define REST routes on server
// 2. Write fetch calls on client
// 3. Manually type the response
// 4. Hope they stay in sync

// tRPC workflow:
// 1. Define a procedure on the server
const appRouter = router({
  getUser: publicProcedure
    .input(z.object({ id: z.string() }))
    .query(({ input }) => {
      return db.user.findUnique({ where: { id: input.id } });
    }),
});

// 2. Call it from the client — fully typed!
const { data } = trpc.getUser.useQuery({ id: "123" });
// data is automatically typed as User | null
// TypeScript error if you pass wrong input shape

使用 Next.js 设置 tRPC

最常见的 tRPC 设置使用 Next.js 的 App Router。以下是从零开始安装和配置的方法。

安装

npm install @trpc/server @trpc/client @trpc/react-query \
  @trpc/next @tanstack/react-query zod superjson

初始化 tRPC 服务器

// src/server/trpc.ts
import { initTRPC, TRPCError } from "@trpc/server";
import superjson from "superjson";
import { ZodError } from "zod";

export const createTRPCContext = async (opts: {
  headers: Headers;
}) => {
  const session = await getServerSession();
  return {
    db: prisma,
    session,
    headers: opts.headers,
  };
};

const t = initTRPC.context<typeof createTRPCContext>().create({
  transformer: superjson,
  errorFormatter({ shape, error }) {
    return {
      ...shape,
      data: {
        ...shape.data,
        zodError:
          error.cause instanceof ZodError
            ? error.cause.flatten()
            : null,
      },
    };
  },
});

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

创建 API 路由处理器（Next.js App Router）

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

const handler = (req: Request) =>
  fetchRequestHandler({
    endpoint: "/api/trpc",
    req,
    router: appRouter,
    createContext: () =>
      createTRPCContext({ headers: req.headers }),
  });

export { handler as GET, handler as POST };

设置客户端提供者

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

export const trpc = createTRPCReact<AppRouter>();

// src/app/providers.tsx
"use client";
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
import { httpBatchLink } from "@trpc/client";
import { trpc } from "@/lib/trpc";
import superjson from "superjson";
import { useState } from "react";

export function TRPCProvider({ children }: { children: React.ReactNode }) {
  const [queryClient] = useState(() => new QueryClient({
    defaultOptions: {
      queries: { staleTime: 5 * 60 * 1000 },
    },
  }));
  const [trpcClient] = useState(() =>
    trpc.createClient({
      links: [
        httpBatchLink({
          url: "/api/trpc",
          transformer: superjson,
        }),
      ],
    })
  );

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

使用 Express 设置 tRPC

tRPC 也可以作为独立的 Express 中间件使用，适用于不使用 Next.js 的情况。

// server.ts
import express from "express";
import * as trpcExpress from "@trpc/server/adapters/express";
import { appRouter } from "./routers/_app";
import { createTRPCContext } from "./trpc";

const app = express();

app.use(
  "/api/trpc",
  trpcExpress.createExpressMiddleware({
    router: appRouter,
    createContext: ({ req, res }) => ({
      db: prisma,
      session: req.session,
    }),
  })
);

app.listen(3000, () => {
  console.log("Server running on port 3000");
});

路由器和过程

路由器将 API 组织成逻辑组。过程是路由器中的单个端点。有三种类型：查询（读取数据）、变更（写入数据）和订阅（实时流）。

// src/server/routers/user.ts
import { z } from "zod";
import { router, publicProcedure, protectedProcedure } from "../trpc";

export const userRouter = router({
  // Query: fetch data (GET-like)
  getById: publicProcedure
    .input(z.object({ id: z.string() }))
    .query(async ({ ctx, input }) => {
      return ctx.db.user.findUnique({
        where: { id: input.id },
      });
    }),

  // Query: list with pagination
  list: publicProcedure
    .input(z.object({
      page: z.number().int().min(1).default(1),
      limit: z.number().int().min(1).max(100).default(20),
    }))
    .query(async ({ ctx, input }) => {
      const { page, limit } = input;
      const [users, total] = await Promise.all([
        ctx.db.user.findMany({
          skip: (page - 1) * limit,
          take: limit,
        }),
        ctx.db.user.count(),
      ]);
      return { users, total, pages: Math.ceil(total / limit) };
    }),

  // Mutation: write data (POST/PUT/DELETE-like)
  update: protectedProcedure
    .input(z.object({
      id: z.string(),
      name: z.string().min(1).max(100).optional(),
      email: z.string().email().optional(),
    }))
    .mutation(async ({ ctx, input }) => {
      return ctx.db.user.update({
        where: { id: input.id },
        data: input,
      });
    }),
});

嵌套路由器

你可以嵌套路由器以创建层次化的 API 结构。随着 API 增长，这保持代码组织有序。

// src/server/routers/_app.ts
import { router } from "../trpc";
import { userRouter } from "./user";
import { postRouter } from "./post";
import { commentRouter } from "./comment";

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

// Export the type for the client
export type AppRouter = typeof appRouter;

// Client usage:
// trpc.user.getById.useQuery({ id: "123" })
// trpc.post.list.useQuery({ page: 1 })
// trpc.comment.create.useMutation()

使用 Zod 进行输入验证

tRPC 与 Zod 集成进行输入验证。定义输入 schema 时，tRPC 在运行时验证传入数据并自动推断过程处理器的 TypeScript 类型。

import { z } from "zod";
import { router, publicProcedure } from "../trpc";

export const productRouter = router({
  create: publicProcedure
    .input(
      z.object({
        name: z.string().min(1, "Name is required").max(200),
        price: z.number().positive("Price must be positive"),
        category: z.enum(["electronics", "clothing", "food"]),
        tags: z.array(z.string()).max(10).default([]),
        metadata: z.record(z.string(), z.unknown()).optional(),
      })
    )
    .mutation(async ({ ctx, input }) => {
      // input is fully typed:
      // {
      //   name: string;
      //   price: number;
      //   category: "electronics" | "clothing" | "food";
      //   tags: string[];
      //   metadata?: Record<string, unknown>;
      // }
      return ctx.db.product.create({ data: input });
    }),
});

复杂输入 Schema

// Reusable input schemas
const PaginationInput = z.object({
  cursor: z.string().nullish(),
  limit: z.number().int().min(1).max(100).default(20),
});

const DateRangeInput = z.object({
  from: z.date(),
  to: z.date(),
}).refine(
  (data) => data.from <= data.to,
  { message: "Start date must be before end date" }
);

const SearchInput = z.object({
  query: z.string().min(1).max(500),
  filters: z.object({
    category: z.enum(["all", "posts", "users"]).default("all"),
    dateRange: DateRangeInput.optional(),
    sortBy: z.enum(["relevance", "date", "popularity"]).default("relevance"),
  }).default({}),
  pagination: PaginationInput.default({}),
});

export const searchRouter = router({
  search: publicProcedure
    .input(SearchInput)
    .query(async ({ ctx, input }) => {
      // All fields are validated and typed
      const { query, filters, pagination } = input;
      return ctx.db.search(query, filters, pagination);
    }),
});

查询和变更

查询获取数据，在客户端用 useQuery 调用。变更修改数据，用 useMutation 调用。两者都完全类型安全。

从客户端调用

// Using queries in React components
function UserProfile({ userId }: { userId: string }) {
  // Fully typed query — data type is inferred from server
  const { data: user, isLoading, error } = trpc.user.getById.useQuery(
    { id: userId },
    { enabled: !!userId }
  );

  // Fully typed mutation
  const updateUser = trpc.user.update.useMutation({
    onSuccess: () => {
      // Invalidate and refetch the query after mutation
      utils.user.getById.invalidate({ id: userId });
    },
    onError: (err) => {
      // err.data?.zodError has field-level errors
      console.error("Update failed:", err.message);
    },
  });

  const utils = trpc.useUtils();

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

  return (
    <form onSubmit={(e) => {
      e.preventDefault();
      updateUser.mutate({
        id: userId,
        name: "New Name",  // TypeScript ensures valid fields
      });
    }}>
      <p>{user.name}</p>
      <p>{user.email}</p>
      <button type="submit" disabled={updateUser.isPending}>
        {updateUser.isPending ? "Saving..." : "Save"}
      </button>
    </form>
  );
}

中间件

中间件让你在过程执行前运行共享逻辑。常见用途包括认证检查、日志记录、速率限制和计时。

认证中间件

// src/server/trpc.ts
const isAuthed = t.middleware(async ({ ctx, next }) => {
  if (!ctx.session?.user) {
    throw new TRPCError({
      code: "UNAUTHORIZED",
      message: "You must be logged in",
    });
  }
  // Add user to context for downstream procedures
  return next({
    ctx: {
      session: ctx.session,
      user: ctx.session.user,  // Now guaranteed to exist
    },
  });
});

export const protectedProcedure = t.procedure.use(isAuthed);

// Logging middleware
const logger = t.middleware(async ({ path, type, next }) => {
  const start = Date.now();
  const result = await next();
  const duration = Date.now() - start;
  console.log(
    `[tRPC] \${type} \${path} - \${duration}ms - ` +
    `\${result.ok ? "OK" : "ERROR"}`
  );
  return result;
});

// Role-based middleware
const isAdmin = t.middleware(async ({ ctx, next }) => {
  if (ctx.session?.user?.role !== "admin") {
    throw new TRPCError({
      code: "FORBIDDEN",
      message: "Admin access required",
    });
  }
  return next({ ctx });
});

export const adminProcedure = protectedProcedure.use(isAdmin);

上下文

上下文是所有过程和中间件可用的数据。它按请求创建，通常包含数据库连接、认证用户和其他请求范围状态。

// src/server/context.ts
import { prisma } from "@/lib/prisma";
import { getServerSession } from "next-auth";
import { authOptions } from "@/lib/auth";

export const createTRPCContext = async (opts: {
  headers: Headers;
}) => {
  const session = await getServerSession(authOptions);

  return {
    // Database client — shared across all procedures
    db: prisma,
    // Authenticated session (null if not logged in)
    session,
    // Request headers for IP, user-agent, etc.
    headers: opts.headers,
    // Request ID for tracing
    requestId: crypto.randomUUID(),
  };
};

export type Context = Awaited<ReturnType<typeof createTRPCContext>>;

// Procedures access context via ctx:
// .query(async ({ ctx }) => {
//   ctx.db       — Prisma client
//   ctx.session  — Auth session
//   ctx.headers  — Request headers
// })

错误处理

tRPC 提供 TRPCError 类，用于抛出带 HTTP 状态码的结构化错误。这些错误被序列化并发送到客户端。

import { TRPCError } from "@trpc/server";

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

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

      // Check permission
      if (post.isPrivate && post.authorId !== ctx.session?.user?.id) {
        throw new TRPCError({
          code: "FORBIDDEN",
          message: "You do not have permission to view this post",
        });
      }

      return post;
    }),

  delete: protectedProcedure
    .input(z.object({ id: z.string() }))
    .mutation(async ({ ctx, input }) => {
      try {
        return await ctx.db.post.delete({
          where: { id: input.id, authorId: ctx.user.id },
        });
      } catch (e) {
        throw new TRPCError({
          code: "INTERNAL_SERVER_ERROR",
          message: "Failed to delete post",
          cause: e,
        });
      }
    }),
});

// Available error codes:
// PARSE_ERROR          400
// BAD_REQUEST          400
// UNAUTHORIZED         401
// FORBIDDEN            403
// NOT_FOUND            404
// METHOD_NOT_SUPPORTED 405
// TIMEOUT              408
// CONFLICT             409
// TOO_MANY_REQUESTS    429
// INTERNAL_SERVER_ERROR 500

自定义错误格式化

// Custom error formatting in initTRPC
const t = initTRPC.context<Context>().create({
  errorFormatter({ shape, error }) {
    return {
      ...shape,
      data: {
        ...shape.data,
        // Include Zod validation errors
        zodError:
          error.cause instanceof ZodError
            ? error.cause.flatten()
            : null,
      },
    };
  },
});

// Client-side error handling
const mutation = trpc.user.update.useMutation({
  onError: (error) => {
    if (error.data?.zodError) {
      // Field-level validation errors
      const fieldErrors = error.data.zodError.fieldErrors;
      // { name: ["Too short"], email: ["Invalid email"] }
    } else {
      // General server error
      toast.error(error.message);
    }
  },
});

React Query 集成

tRPC 包装 React Query（TanStack Query）提供类型安全的数据获取 hooks。你可以获得所有 React Query 功能。

乐观更新

function TodoList() {
  const utils = trpc.useUtils();
  const todos = trpc.todo.list.useQuery();

  const toggleTodo = trpc.todo.toggle.useMutation({
    // Optimistically update the UI before server responds
    onMutate: async (input) => {
      // Cancel any outgoing refetches
      await utils.todo.list.cancel();

      // Snapshot the previous value
      const previousTodos = utils.todo.list.getData();

      // Optimistically update the cache
      utils.todo.list.setData(undefined, (old) =>
        old?.map((t) =>
          t.id === input.id
            ? { ...t, completed: !t.completed }
            : t
        )
      );

      return { previousTodos };
    },
    // If mutation fails, roll back to snapshot
    onError: (err, input, context) => {
      utils.todo.list.setData(undefined, context?.previousTodos);
    },
    // Always refetch after error or success
    onSettled: () => {
      utils.todo.list.invalidate();
    },
  });

  return (
    <ul>
      {todos.data?.map((todo) => (
        <li key={todo.id} onClick={() => toggleTodo.mutate({ id: todo.id })}>
          {todo.completed ? "[x]" : "[ ]"} {todo.text}
        </li>
      ))}
    </ul>
  );
}

无限查询

// Server: cursor-based pagination
export const feedRouter = router({
  infiniteFeed: publicProcedure
    .input(z.object({
      cursor: z.string().nullish(),
      limit: z.number().min(1).max(50).default(20),
    }))
    .query(async ({ ctx, input }) => {
      const items = await ctx.db.post.findMany({
        take: input.limit + 1,
        cursor: input.cursor ? { id: input.cursor } : undefined,
        orderBy: { createdAt: "desc" },
      });
      let nextCursor: string | undefined;
      if (items.length > input.limit) {
        const nextItem = items.pop();
        nextCursor = nextItem?.id;
      }
      return { items, nextCursor };
    }),
});

// Client: infinite scroll
function Feed() {
  const feed = trpc.feed.infiniteFeed.useInfiniteQuery(
    { limit: 20 },
    { getNextPageParam: (lastPage) => lastPage.nextCursor }
  );

  return (
    <div>
      {feed.data?.pages.map((page) =>
        page.items.map((item) => (
          <PostCard key={item.id} post={item} />
        ))
      )}
      <button
        onClick={() => feed.fetchNextPage()}
        disabled={!feed.hasNextPage || feed.isFetchingNextPage}
      >
        {feed.isFetchingNextPage ? "Loading..." : "Load more"}
      </button>
    </div>
  );
}

订阅（实时）

订阅使用 WebSocket 将数据从服务器实时推送到客户端。适用于聊天、实时仪表板、通知等。

// Server: WebSocket subscription
import { observable } from "@trpc/server/observable";
import { EventEmitter } from "events";

const ee = new EventEmitter();

export const chatRouter = router({
  sendMessage: protectedProcedure
    .input(z.object({
      roomId: z.string(),
      text: z.string().min(1).max(1000),
    }))
    .mutation(async ({ ctx, input }) => {
      const message = {
        id: crypto.randomUUID(),
        text: input.text,
        roomId: input.roomId,
        userId: ctx.user.id,
        createdAt: new Date(),
      };
      await ctx.db.message.create({ data: message });
      ee.emit("newMessage", message);
      return message;
    }),

  onNewMessage: publicProcedure
    .input(z.object({ roomId: z.string() }))
    .subscription(({ input }) => {
      return observable((emit) => {
        const handler = (message: Message) => {
          if (message.roomId === input.roomId) {
            emit.next(message);
          }
        };
        ee.on("newMessage", handler);
        return () => ee.off("newMessage", handler);
      });
    }),
});

// Client: subscribe to messages
function ChatRoom({ roomId }: { roomId: string }) {
  trpc.chat.onNewMessage.useSubscription(
    { roomId },
    {
      onData: (message) => {
        // Append new message to local state
        setMessages((prev) => [...prev, message]);
      },
      onError: (err) => {
        console.error("Subscription error:", err);
      },
    }
  );
}

测试 tRPC 过程

你可以直接调用路由器来测试 tRPC 过程，无需 HTTP 层。这使单元测试快速而简单。

// __tests__/user.test.ts
import { describe, it, expect, beforeEach } from "vitest";
import { appRouter } from "@/server/routers/_app";
import { createCallerFactory } from "@/server/trpc";
import { prisma } from "@/lib/__mocks__/prisma";

const createCaller = createCallerFactory(appRouter);

describe("user router", () => {
  it("should return a user by ID", async () => {
    const caller = createCaller({
      db: prisma,
      session: null,
      headers: new Headers(),
    });

    prisma.user.findUnique.mockResolvedValue({
      id: "1",
      name: "Alice",
      email: "alice@example.com",
    });

    const user = await caller.user.getById({ id: "1" });
    expect(user).toEqual({
      id: "1",
      name: "Alice",
      email: "alice@example.com",
    });
  });

  it("should throw NOT_FOUND for missing user", async () => {
    const caller = createCaller({
      db: prisma,
      session: null,
      headers: new Headers(),
    });

    prisma.user.findUnique.mockResolvedValue(null);

    await expect(
      caller.user.getById({ id: "nonexistent" })
    ).rejects.toThrow("NOT_FOUND");
  });

  it("should reject invalid input", async () => {
    const caller = createCaller({
      db: prisma,
      session: { user: { id: "1", role: "user" } },
      headers: new Headers(),
    });

    await expect(
      caller.user.update({ id: "1", email: "not-an-email" })
    ).rejects.toThrow();
  });

  it("should require auth for protected routes", async () => {
    const caller = createCaller({
      db: prisma,
      session: null,  // Not logged in
      headers: new Headers(),
    });

    await expect(
      caller.user.update({ id: "1", name: "Hacker" })
    ).rejects.toThrow("UNAUTHORIZED");
  });
});

性能优化技巧

tRPC 本身已经很快，但还有额外的优化可以应用。

使用 HTTP 批处理：tRPC 默认将多个请求批处理为单个 HTTP 调用
启用转换器：使用 superjson 序列化 Date、Map、Set 等非 JSON 类型
利用 React Query 缓存：配置 staleTime 和 gcTime 避免不必要的重新获取
使用服务端调用：在 Next.js Server Components 中直接调用过程
拆分路由器：对大型路由器使用动态导入以减少包大小
连接池：在上下文创建中重用数据库连接

// Server-side calls in Next.js Server Components
// (no HTTP overhead — direct function call)
import { createCallerFactory } from "@/server/trpc";
import { appRouter } from "@/server/routers/_app";

const createCaller = createCallerFactory(appRouter);

export default async function UserPage({
  params,
}: {
  params: { id: string };
}) {
  const caller = createCaller(await createTRPCContext({
    headers: headers(),
  }));

  // Direct procedure call — no HTTP request
  const user = await caller.user.getById({ id: params.id });

  return <UserProfileView user={user} />;
}

常见问题

tRPC 与 REST 和 GraphQL 有何不同？

REST 需要手动维护 API 契约且没有内置类型安全。GraphQL 提供 schema 但需要代码生成获取 TypeScript 类型。tRPC 通过直接使用 TypeScript 推断跳过两者。缺点是 tRPC 仅适用于 TypeScript 客户端。

能否将 tRPC 与非 TypeScript 客户端一起使用？

tRPC 专为 TypeScript 到 TypeScript 通信设计。对于非 TypeScript 客户端，可以使用 trpc-openapi 生成 OpenAPI 规范。

tRPC 是否取代 React Query？

不是。tRPC 基于 React Query 构建，在其之上添加类型安全的过程调用。

如何用 tRPC 处理文件上传？

tRPC 不原生处理文件上传。可以使用单独的 REST 端点或预签名 URL 方式。

不使用 Next.js 能用 tRPC 吗？

可以。tRPC 有 Express、Fastify、AWS Lambda、Cloudflare Workers 等适配器。

tRPC 可以用于生产吗？

可以。tRPC v11 稳定且被各种规模的公司用于生产环境。

如何对 tRPC API 进行版本管理？

在 TypeScript monorepo 中，版本管理由部署流程处理。对于外部消费者，使用 trpc-openapi 暴露版本化的 REST 端点。

tRPC 支持服务端渲染吗？

支持。在 Next.js 中，可以使用 createCaller API 或 React Server Components 在服务器上预获取查询。