MCP Tools for Pothos GraphQL — scope auth, complexity plugin, Prisma N+1 protection

Scope auth plugin — ForbiddenError and GraphQL partial responses

Pothos's @pothos/plugin-scope-auth integrates authorization at the schema level. Each type and field can declare required scopes via authScopes. When a scope check fails, the plugin throws a ForbiddenError before the resolver runs. GraphQL execution catches this error, sets the field to null, and adds the error to the errors array in the response — HTTP status remains 200.

import SchemaBuilder from '@pothos/core';
import ScopeAuthPlugin from '@pothos/plugin-scope-auth';

// Define your auth scopes — these are evaluated per-request
const builder = new SchemaBuilder<{
  Context: {
    viewer: { id: string; role: 'user' | 'admin' } | null;
  };
  AuthScopes: {
    isAuthenticated: boolean;
    isAdmin: boolean;
    canEditPost: { postId: string };
  };
}>({
  plugins: [ScopeAuthPlugin],
  scopeAuth: {
    // authScopes is called once per request to build the scope set
    authScopes: async (ctx) => ({
      isAuthenticated: !!ctx.viewer,
      isAdmin: ctx.viewer?.role === 'admin',
      // canEditPost requires dynamic scope check: author check happens here
      canEditPost: async (required) => {
        if (!ctx.viewer) return false;
        const post = await ctx.prisma.post.findUnique({
          where: { id: required.postId },
          select: { authorId: true },
        });
        return post?.authorId === ctx.viewer.id;
      },
    }),
  },
});

// Apply scopes at the field level
builder.queryField('myProfile', (t) =>
  t.prismaField({
    type: 'User',
    // This field requires isAuthenticated scope
    // If context.viewer is null, ForbiddenError is thrown → field is null in response
    authScopes: { isAuthenticated: true },
    resolve: (query, _root, _args, ctx) =>
      ctx.prisma.user.findUniqueOrThrow({
        ...query,
        where: { id: ctx.viewer!.id },
      }),
  })
);

builder.mutationField('deletePost', (t) =>
  t.field({
    type: 'Boolean',
    args: { postId: t.arg.string({ required: true }) },
    // Dynamic scope: checks if viewer is the post author
    authScopes: (_root, args) => ({ canEditPost: { postId: args.postId } }),
    resolve: async (_root, args, ctx) => {
      await ctx.prisma.post.delete({ where: { id: args.postId } });
      return true;
    },
  })
);

// MCP tool client: handle partial responses from scope auth failures
async function callGraphQL(query: string, variables: Record<string, unknown>, token: string) {
  const response = await fetch(GRAPHQL_URL, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json', 'Authorization': \`Bearer \${token}\` },
    body: JSON.stringify({ query, variables }),
  });

  const body = await response.json();

  // HTTP 200 even on auth failure — check errors array
  if (body.errors?.length) {
    const authError = body.errors.find((e: { extensions?: { code?: string } }) =>
      e.extensions?.code === 'FORBIDDEN'
    );
    if (authError) {
      throw new Error(\`Authorization denied: \${authError.message}\`);
    }
    throw new Error(body.errors[0].message);
  }

  return body.data;
}

Scope functions can be async — they receive the resolved scope requirement object as their argument. For expensive scope checks (like database lookups), Pothos caches the result per-request: if a scope is checked multiple times in one request (multiple fields with the same scope), the function is only called once. This makes field-level authorization practical even for database-backed permission checks.

Complexity plugin — list multipliers and per-field cost

The @pothos/plugin-complexity plugin counts the cost of a GraphQL query before executing it. Without a list multiplier, a query for 1000 users costs the same as a query for 1 user — the plugin counts fields, not field-times-cardinality. Setting a multiplier based on pagination arguments makes the cost budget reflect actual database load.

import ComplexityPlugin from '@pothos/plugin-complexity';

const builder = new SchemaBuilder<{ Context: AppContext }>({
  plugins: [ScopeAuthPlugin, ComplexityPlugin],
  complexity: {
    // Maximum total complexity per query
    limit: { complexity: 500, depth: 8, breadth: 15 },

    // Default cost for scalar fields (ID, String, Int, etc.)
    defaultComplexity: 1,

    // Default multiplier for list fields when no explicit multiplier is set
    // 10 means: each element in a list costs 10x the field's base cost
    // Override per field using the complexity option on field builders
    defaultListMultiplier: 10,
  },
});

builder.queryField('users', (t) =>
  t.prismaField({
    type: ['User'],
    args: {
      first: t.arg.int({ defaultValue: 20 }),
      after: t.arg.string(),
    },
    // Field-specific complexity: reads the first argument to scale cost
    complexity: (args, childComplexity) =>
      // Each user costs childComplexity (sum of its selected fields)
      // Multiplied by the number of users requested
      (args.first ?? 20) * childComplexity + 1,
    resolve: (query, _root, args, ctx) =>
      ctx.prisma.user.findMany({
        ...query,
        take: args.first ?? 20,
        cursor: args.after ? { id: args.after } : undefined,
      }),
  })
);

builder.objectField(builder.objectRef<User>('User'), 'posts', (t) =>
  t.prismaField({
    type: ['Post'],
    args: { first: t.arg.int({ defaultValue: 10 }) },
    // Nested list: also scales by cardinality
    complexity: (args, childComplexity) =>
      (args.first ?? 10) * childComplexity + 1,
    resolve: (query, user, args, ctx) =>
      ctx.prisma.post.findMany({
        ...query,
        where: { authorId: user.id },
        take: args.first ?? 10,
      }),
  })
);

// With these settings:
// query { users(first: 100) { id name posts(first: 50) { id title } } }
// complexity = 100 * (1 + 1 + 50 * (1 + 1)) + 1 = 100 * 102 + 1 = 10_201
// This exceeds the limit of 500 — rejected before execution
//
// query { users(first: 5) { id name } }
// complexity = 5 * (1 + 1) + 1 = 11 — allowed

For MCP tool servers where queries are constructed by an LLM, set the complexity limit conservatively. LLMs don't have intuition about cost — they may generate queries with large first arguments because larger datasets look more "complete" in a reasoning context. A limit of 500 with a defaultListMultiplier of 10 rejects most inadvertently expensive queries while allowing typical tool-call patterns.

Prisma plugin — N+1 protection and when it applies

The @pothos/plugin-prisma plugin uses Prisma's include and select capabilities to batch relation loads. When you declare relations using t.relation(), the plugin automatically joins the related data in a single Prisma query rather than issuing one query per parent. This is fundamentally different from DataLoader batching — instead of grouping queries, it avoids extra queries entirely by letting Prisma fetch nested data in one call.

import PrismaPlugin from '@pothos/plugin-prisma';
import type PrismaTypes from '@pothos/plugin-prisma/generated';

const builder = new SchemaBuilder<{
  Context: { prisma: PrismaClient; viewer: { id: string } | null };
  PrismaTypes: PrismaTypes;
}>({
  plugins: [PrismaPlugin],
  prisma: {
    client: (ctx) => ctx.prisma,
  },
});

// WRONG: calling prisma directly inside a resolver bypasses N+1 protection
builder.prismaObject('User', {
  fields: (t) => ({
    id: t.exposeID('id'),
    name: t.exposeString('name'),
    postsManual: t.field({
      type: ['Post'],
      // WRONG: one prisma.post.findMany() call per user — N+1 pattern
      resolve: async (user, _args, ctx) =>
        ctx.prisma.post.findMany({ where: { authorId: user.id } }),
    }),
  }),
});

// RIGHT: use t.relation() — Pothos tells Prisma to include posts in the parent query
builder.prismaObject('User', {
  fields: (t) => ({
    id: t.exposeID('id'),
    name: t.exposeString('name'),
    // t.relation() — no extra query; posts fetched with the user query via Prisma include
    posts: t.relation('posts', {
      args: { first: t.arg.int({ defaultValue: 10 }) },
      query: (args) => ({ take: args.first ?? 10, orderBy: { createdAt: 'desc' } }),
    }),
  }),
});

// The prismaField on queries passes a 'query' argument to tell Prisma what to include
builder.queryField('user', (t) =>
  t.prismaField({
    type: 'User',
    nullable: true,
    args: { id: t.arg.id({ required: true }) },
    resolve: (query, _root, args, ctx) =>
      // 'query' contains the Prisma include/select that the Pothos plugin computed
      // from the GraphQL selection set — pass it through to prisma
      ctx.prisma.user.findUnique({
        ...query,          // includes posts (and nested relations) if selected
        where: { id: String(args.id) },
      }),
  })
);

// IMPORTANT: the 'query' argument MUST be spread into the Prisma call
// Omitting ...query causes the Prisma plugin's relation batching to break
// and falls back to separate queries per relation field (N+1 returns)

// Verify N+1 protection is working by enabling Prisma query logging:
const prismaWithLogging = new PrismaClient({
  log: [{ emit: 'event', level: 'query' }],
});
prismaWithLogging.$on('query', (e) => {
  console.log('Prisma query:', e.query);
  // If you see dozens of identical SELECT statements per request → N+1 problem
  // If you see one JOIN-based SELECT with nested data → Pothos plugin is working
});

Health probe — Pothos GraphQL reachability from an MCP server

A Pothos schema is served through a standard GraphQL server (Apollo Server, Yoga, Mercurius, etc.). The health probe targets the GraphQL execution endpoint. The { __typename } query is always valid and requires no auth — it exercises schema parsing, validation, and execution without touching the database or auth system.

async function checkPothosHealth(graphqlUrl: string): Promise<{
  healthy: boolean;
  latencyMs?: number;
  error?: string;
}> {
  const start = Date.now();
  try {
    const response = await fetch(graphqlUrl, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ query: '{ __typename }' }),
      signal: AbortSignal.timeout(5_000),
    });

    if (!response.ok) {
      return { healthy: false, error: \`HTTP \${response.status}\` };
    }

    const body = await response.json();

    // Pothos complexity limit errors surface here (before execution)
    // { errors: [{ message: 'Query complexity limit exceeded', extensions: { code: 'COMPLEXITY_LIMIT_EXCEEDED' } }] }
    if (body.errors) {
      return { healthy: false, error: body.errors[0]?.message };
    }

    return { healthy: true, latencyMs: Date.now() - start };
  } catch (err) {
    return { healthy: false, error: String(err) };
  }
}

// Deeper probe: check scope auth is wired correctly by requesting an auth-gated field
async function checkAuthProbe(graphqlUrl: string, token: string): Promise<{
  authWorking: boolean;
  error?: string;
}> {
  const response = await fetch(graphqlUrl, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Authorization': \`Bearer \${token}\`,
    },
    body: JSON.stringify({ query: '{ myProfile { id } }' }),
    signal: AbortSignal.timeout(5_000),
  });

  const body = await response.json();

  // Scope auth failure: HTTP 200 with errors array, data.myProfile: null
  const hasForbidden = body.errors?.some(
    (e: { extensions?: { code?: string } }) => e.extensions?.code === 'FORBIDDEN'
  );

  return {
    authWorking: !hasForbidden && body.data?.myProfile?.id != null,
    error: hasForbidden ? 'Scope auth rejected probe token' : undefined,
  };
}

AliveMCP monitors Pothos GraphQL endpoints with both the structural probe (__typename) and an authenticated probe to catch scope-auth regressions — a Pothos deploy that accidentally removes a required scope or wires the wrong context property causes the auth probe to silently fail with ForbiddenErrors while the basic health check still returns 200.