MCP Server AppSync Subscriptions — real-time tool result streaming via GraphQL

Schema design: @aws_subscribe and subscription filters

The simplest AppSync subscription pattern requires no resolver: add the @aws_subscribe directive to the subscription field and list the mutation fields that should trigger it. AppSync automatically applies argument-based filtering — if a subscriber passes sessionId: "abc", they only receive events where the mutation response contains sessionId: "abc".

type ToolResult {
  toolCallId: ID!
  sessionId: ID!
  toolName: String!
  status: String!       # "running" | "complete" | "error"
  output: String        # JSON-serialized result, null while running
  durationMs: Int
  createdAt: AWSDateTime!
}

type Mutation {
  createToolResult(input: ToolResultInput!): ToolResult!
    @aws_api_key @aws_cognito_user_pools
}

type Subscription {
  # Clients subscribe with their sessionId — AppSync filters server-side
  onToolResult(sessionId: ID!): ToolResult
    @aws_subscribe(mutations: ["createToolResult"])
    @aws_api_key @aws_cognito_user_pools
}

# AppSync subscription filter: the argument value is matched against
# the field of the same name in the mutation response object.
# If the subscriber passes sessionId: "abc", AppSync only delivers
# events where the createToolResult response.sessionId === "abc".

If you need more complex filter logic — filtering on a nested field, combining multiple conditions with AND/OR, or applying a filter not expressible as a direct field match — use the enhanced subscription filter API introduced in 2023. In a Lambda or JavaScript subscription resolver, return a filterGroup in the resolver response that AppSync evaluates against each incoming mutation payload.

// JavaScript subscription resolver (AppSync JS runtime) for complex filters
// This runs when a client subscribes, not when the event fires
export function request(ctx) {
  // ctx.args contains the subscription arguments the client passed
  const { sessionId, toolNames } = ctx.args;
  return {
    payload: null  // no initial payload for subscriptions
  };
}

export function response(ctx) {
  const { sessionId, toolNames } = ctx.args;
  // Return an enhanced filter that AppSync evaluates on each mutation event
  const filter = {
    filterGroup: [
      {
        filters: [
          { fieldName: "sessionId", operator: "eq", value: sessionId },
          // Only deliver events for specific tool names (optional)
          ...(toolNames
            ? [{ fieldName: "toolName", operator: "in", value: toolNames }]
            : [])
        ]
      }
    ]
  };
  extensions.setSubscriptionFilter(filter);
  return null;
}

Lambda mutation resolver: the shape that triggers subscriptions

When using a Lambda data source for the createToolResult mutation, the Lambda must return an object whose shape exactly matches the ToolResult GraphQL type — including all non-nullable fields. If a required field is missing, AppSync will return an error to the mutation caller and no subscription events will fire. The Lambda does not write to DynamoDB itself in the simplest pattern — it writes the result object to DynamoDB using a pipeline function, and the final function returns the saved item. However, a simpler unit resolver that returns the item directly also works.

// Lambda handler for createToolResult mutation
// AppSync sends a batch request when using Lambda data source with batching enabled
export const handler = async (event) => {
  // event.arguments contains the mutation input
  // event.identity contains the caller's identity (Cognito claims or API key)
  const { input } = event.arguments;

  const result = {
    toolCallId: input.toolCallId,
    sessionId: input.sessionId,
    toolName: input.toolName,
    status: input.status,
    output: input.output ?? null,   // null is valid for non-nullable String? (optional)
    durationMs: input.durationMs ?? null,
    createdAt: new Date().toISOString()
  };

  // Persist to DynamoDB directly from the Lambda (simpler than a pipeline)
  await dynamoDb.put({
    TableName: process.env.TABLE_NAME,
    Item: result
  }).promise();

  // Return the exact shape of the ToolResult GraphQL type.
  // AppSync matches this against active subscriptions and fans out to
  // subscribers where filter conditions match.
  return result;
};

// IMPORTANT: Lambda response for a mutation that also has a subscription:
// - All non-nullable fields must be present and non-null
// - AWSDateTime fields must be ISO 8601 format: "2026-09-18T10:00:00.000Z"
// - ID fields must be strings (AppSync does NOT coerce numbers to strings)
// - Extra fields not in the schema are silently ignored

When using AppSync's NONE data source for a local resolver (no backend call — useful for real-time pub/sub without persistence), the mutation resolver simply passes the input through to subscribers without writing to any backend. This is the lowest-latency pattern for streaming partial tool results where persistence is handled separately.

// NONE data source JavaScript resolver — pass-through mutation
// Useful for streaming partial updates (SSE-equivalent via AppSync)
export function request(ctx) {
  return { payload: ctx.args.input };
}

export function response(ctx) {
  // ctx.result is the payload returned from request()
  return ctx.result;
}

Client connection lifecycle: timeouts, reconnection, and keep-alive

AppSync WebSocket connections operate over MQTT using the wss:// endpoint at realtime.appsync-realtime-api.{region}.amazonaws.com (not the regular HTTPS endpoint). The connection lifecycle has three phases: HTTP upgrade (with auth headers in the URL as base64-encoded JSON), MQTT CONNECT, and subscribe. Idle connections are terminated after 2 hours; after receiving the last subscription event, a connection that sends no messages will be closed at the 2-hour mark.

// Using @aws-amplify/api-graphql for subscriptions (browser + Node)
import { generateClient } from 'aws-amplify/api';

const client = generateClient();

// Subscribe to tool results for this session
const sub = client.graphql({
  query: `subscription OnToolResult($sessionId: ID!) {
    onToolResult(sessionId: $sessionId) {
      toolCallId
      sessionId
      toolName
      status
      output
      durationMs
      createdAt
    }
  }`,
  variables: { sessionId: currentSessionId }
}).subscribe({
  next: ({ data }) => {
    const result = data.onToolResult;
    handleToolResult(result);
  },
  error: (err) => {
    // AppSync subscription errors arrive here — includes connection drops
    console.error('Subscription error', err);
    // Amplify auto-reconnects with exponential backoff (up to 5 min)
    // No manual reconnect needed for transient errors
  }
});

// Clean up when the MCP session ends
function endSession() {
  sub.unsubscribe();
}

// For server-side Node.js (no Amplify), use aws-appsync package:
// import AWSAppSyncClient from 'aws-appsync';
// AWSAppSyncClient handles MQTT, auth, and reconnection automatically

When using raw WebSocket (without Amplify), you must send a ping to the AppSync endpoint every 300 seconds — AppSync sends {"type":"ka"} keep-alive frames, but the client is responsible for responding and for detecting connection loss via a connection_keep_alive_timeout message. If the client receives a connection_keep_alive_timeout and has not sent a ping within 300 seconds, AppSync closes the connection. Mobile apps that background the app must re-establish the WebSocket connection on foreground resume.

Authorization on subscription fields

AppSync evaluates authorization on subscription fields at two points: when the client subscribes (connection-time auth) and — optionally, with enhanced subscriptions — per-event auth. The authorization modes available for subscription fields are the same as for queries and mutations: API key, Amazon Cognito user pools, IAM, Lambda authorizer, and OpenID Connect. The @aws_subscribe directive does not inherit auth from the linked mutation — each subscription field must have its own auth directive.

// Schema with multiple auth modes on subscription field
type Subscription {
  # Allow both API key (public dashboard) and Cognito (authenticated users)
  onToolResult(sessionId: ID!): ToolResult
    @aws_subscribe(mutations: ["createToolResult"])
    @aws_api_key
    @aws_cognito_user_pools

  # Private subscription — Cognito users only, additionally filtered by userId claim
  onPrivateToolResult(sessionId: ID!): ToolResult
    @aws_subscribe(mutations: ["createPrivateToolResult"])
    @aws_cognito_user_pools
}

// Lambda authorizer for AppSync (applies at connection time, not per-event)
// Returns { isAuthorized: true, resolverContext: { userId: "..." }, ttlOverride: 300 }
export const handler = async (event) => {
  const { authorizationToken, requestContext } = event;
  // Validate JWT, API key, etc.
  const claims = verifyToken(authorizationToken);
  return {
    isAuthorized: !!claims,
    resolverContext: {
      userId: claims?.sub ?? null
    },
    ttlOverride: 300  // cache this authorization result for 5 minutes
  };
};

Failure modes reference