Category: JavaScript

ANTHROPIC_API_KEY=sk-ant-xxxxxxxx

En `app.module.ts`, registra `ConfigModule`:

```typescript
// src/app.module.ts
import { Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config';
import { AiModule } from './ai/ai.module';

@Module({
  imports: [
    ConfigModule.forRoot({ isGlobal: true }),
    AiModule,
  ],
})
export class AppModule {}

isGlobal: true significa que ConfigService está disponible en todos los módulos sin importarlo individualmente. Práctico.

La estructura del AiModule

Antes de escribir código, la estructura:

src/
  ai/
    ai.module.ts
    ai.controller.ts
    ai.service.ts
    dto/
      chat.dto.ts

Cuatro archivos. Eso es todo lo que necesita un endpoint de streaming limpio.

Paso 1: El DTO de validación

El primer punto de defensa es el DTO. Define el contrato del request:

// src/ai/dto/chat.dto.ts
import { IsArray, IsIn, IsString, ValidateNested, ArrayMinSize } from 'class-validator';
import { Type } from 'class-transformer';

export class ChatMessageDto {
  @IsIn(['user', 'assistant', 'system'])
  role: 'user' | 'assistant' | 'system';

  @IsString()
  content: string;
}

export class ChatRequestDto {
  @IsArray()
  @ArrayMinSize(1)
  @ValidateNested({ each: true })
  @Type(() => ChatMessageDto)
  messages: ChatMessageDto[];
}

@ValidateNested({ each: true }) valida cada elemento del array individualmente. Si el frontend manda un mensaje con role: 'hacker' o sin content, el request rebota antes de tocar el servicio.

Para que ValidationPipe funcione globalmente, añádelo en main.ts:

// src/main.ts
import { NestFactory } from '@nestjs/core';
import { ValidationPipe } from '@nestjs/common';
import { AppModule } from './app.module';

async function bootstrap() {
  const app = NestFactory.create(AppModule);

  app.useGlobalPipes(new ValidationPipe({
    transform: true,
    whitelist: true,    // elimina propiedades no declaradas en el DTO
    forbidNonWhitelisted: true,
  }));

  // CORS para el frontend Angular en desarrollo
  app.enableCors({
    origin: process.env.FRONTEND_URL ?? 'http://localhost:4200',
    methods: ['POST', 'OPTIONS'],
  });

  await app.listen(process.env.PORT ?? 3000);
}

bootstrap();

whitelist: true es especialmente importante aquí: elimina cualquier campo del body que no esté declarado en el DTO. Si alguien intenta inyectar propiedades extra en el request, NestJS las ignora antes de que lleguen al servicio.

Paso 2: El AiService

El servicio encapsula toda la lógica de llamada al modelo. El controlador no sabe qué modelo usamos ni cómo se configura — solo llama al servicio y recibe el stream.

// src/ai/ai.service.ts
import { Injectable } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { streamText, CoreMessage } from 'ai';
import { createAnthropic } from '@ai-sdk/anthropic';

@Injectable()
export class AiService {
  private readonly anthropic;

  constructor(private readonly config: ConfigService) {
    this.anthropic = createAnthropic({
      apiKey: this.config.getOrThrow<string>('ANTHROPIC_API_KEY'),
    });
  }

  streamChat(messages: CoreMessage[]) {
    return streamText({
      model: this.anthropic('claude-sonnet-4-6'),
      system: `Eres un asistente técnico especializado en desarrollo de software.
Responde en español de forma concisa y directa.
Si el usuario pregunta sobre código, incluye ejemplos concretos.`,
      messages,
      maxTokens: 1024,
    });
  }
}

Dos decisiones importantes aquí:

createAnthropic({ apiKey }) en el constructor — el cliente de Anthropic se crea una sola vez cuando NestJS instancia el servicio. No se recrea en cada petición. Eso evita overhead innecesario.

config.getOrThrow<string>('ANTHROPIC_API_KEY') — si la variable de entorno no existe, la app falla en el arranque con un error claro en lugar de fallar silenciosamente en el primer request. Fail fast.

maxTokens: 1024 es un límite defensivo. Sin él, un usuario puede hacer una pregunta que genere una respuesta de 8.000 tokens, multiplicando el costo por 8. Ajusta según tu caso de uso.

Paso 3: El AiController con streaming

El controlador es donde ocurre la magia del streaming. La clave está en cómo NestJS maneja la respuesta HTTP nativa:

// src/ai/ai.controller.ts
import {
  Controller,
  Post,
  Body,
  Res,
  HttpCode,
  HttpStatus,
} from '@nestjs/common';
import { Response } from 'express';
import { AiService } from './ai.service';
import { ChatRequestDto } from './dto/chat.dto';
import { CoreMessage } from 'ai';

@Controller('api')
export class AiController {
  constructor(private readonly aiService: AiService) {}

  @Post('chat')
  @HttpCode(HttpStatus.OK)
  async chat(
    @Body() body: ChatRequestDto,
    @Res() res: Response,
  ): Promise<void> {
    const messages = body.messages as CoreMessage[];

    const result = this.aiService.streamChat(messages);

    // toUIMessageStreamResponse() genera una Response Web estándar
    // con el protocolo SSE del AI SDK
    const streamResponse = result.toUIMessageStreamResponse();

    // Propagamos los headers del AI SDK a la respuesta de Express
    streamResponse.headers.forEach((value, key) => {
      res.setHeader(key, value);
    });

    res.status(streamResponse.status);

    // Volcamos el body del ReadableStream a la respuesta de Express
    if (streamResponse.body) {
      const reader = streamResponse.body.getReader();

      const pump = async () => {
        while (true) {
          const { done, value } = await reader.read();
          if (done) {
            res.end();
            break;
          }
          res.write(value);
        }
      };

      pump().catch((err) => {
        console.error('[AiController] Error en stream:', err);
        if (!res.headersSent) {
          res.status(500).json({ error: 'Error interno del stream' });
        } else {
          res.end();
        }
      });
    } else {
      res.status(500).json({ error: 'No se pudo iniciar el stream' });
    }
  }
}

¿Por qué este patrón de pump manual en lugar de pipe()?

toUIMessageStreamResponse() devuelve una Response Web estándar (la del spec WHATWG), no un stream de Node.js. Express trabaja con streams de Node.js. El pump manual convierte uno en el otro sin dependencias adicionales. Es verboso pero explícito — sabes exactamente qué hace cada línea.

El bloque catch en el pump gestiona dos escenarios: si el error ocurre antes de enviar headers, devuelve un 500 con JSON. Si ocurre después (cuando el stream ya está activo), llama a res.end() para cerrar la conexión limpiamente. Sin este manejo, el cliente se quedaría esperando indefinidamente.

Paso 4: El AiModule

El módulo agrupa las tres piezas:

// src/ai/ai.module.ts
import { Module } from '@nestjs/common';
import { AiController } from './ai.controller';
import { AiService } from './ai.service';

@Module({
  controllers: [AiController],
  providers: [AiService],
  exports: [AiService], // por si otros módulos necesitan AiService
})
export class AiModule {}

Exportar AiService es una decisión de diseño: si en el futuro un módulo de AgentsModule o DocumentModule necesita llamar al modelo, importan AiModule y tienen el servicio disponible sin duplicar configuración.

Rate limiting: el paso que nadie incluye

Sin rate limiting, un solo usuario puede vaciar tu cuota de Anthropic en minutos. NestJS tiene @nestjs/throttler para esto:

npm install @nestjs/throttler

Configúralo en AppModule:

// src/app.module.ts
import { ThrottlerModule, ThrottlerGuard } from '@nestjs/throttler';
import { APP_GUARD } from '@nestjs/core';

@Module({
  imports: [
    ConfigModule.forRoot({ isGlobal: true }),
    ThrottlerModule.forRoot([{
      name: 'short',
      ttl: 60_000,   // 1 minuto en ms
      limit: 10,     // máximo 10 requests por minuto por IP
    }]),
    AiModule,
  ],
  providers: [
    {
      provide: APP_GUARD,
      useClass: ThrottlerGuard,
    },
  ],
})
export class AppModule {}

10 requests por minuto por IP es un límite conservador para un chat. En producción, ajusta según el plan de Anthropic que tengas y el perfil de uso esperado. Si tus usuarios son developers que mandan snippets de código largos, 10 puede ser demasiado restrictivo. Si es un chat de soporte con usuarios anónimos, puede ser demasiado permisivo.

ThrottlerGuard como APP_GUARD aplica el límite a todos los endpoints automáticamente. Si quieres excluir algunos endpoints del límite, usa el decorador @SkipThrottle() en el controlador correspondiente.

Conectar con el frontend Angular

Este backend está diseñado para ser el complemento del post Angular v22 + Vercel AI SDK: streaming de IA en tu app en 20 minutos.

El frontend Angular usa fetch nativo con ReadableStream. El cambio que necesitas en el componente Angular es mínimo: actualizar la URL del endpoint del servidor Bun del post anterior (típicamente en el puerto 4000) a http://localhost:3000/api/chat de este servidor NestJS. El contrato del API no cambia — misma ruta, mismo formato de mensajes.

La diferencia está en el protocolo de stream. El servidor Bun del post anterior usa toTextStreamResponse(), que devuelve texto plano. Este NestJS usa toUIMessageStreamResponse(), que usa el protocolo SSE estructurado del AI SDK. Para consumir este protocolo desde Angular sin la librería useChat de React, el componente Angular necesita parsear los chunks SSE en lugar de concatenarlos directamente.

Si ya tienes el frontend del post anterior y quieres migrar a este backend sin tocar el componente, cambia en AiService.streamChat() el retorno a toTextStreamResponse():

// AiService — variante compatible con el componente Angular del post anterior
streamChat(messages: CoreMessage[]) {
  return streamText({
    model: this.anthropic('claude-sonnet-4-6'),
    system: 'Eres un asistente técnico...',
    messages,
    maxTokens: 1024,
  });
  // En el controlador usar toTextStreamResponse() en vez de toUIMessageStreamResponse()
}

Y en el controlador, sustituye result.toUIMessageStreamResponse() por result.toTextStreamResponse(). El componente Angular del post anterior funciona sin cambios.

La versión con toUIMessageStreamResponse() es la recomendada para proyectos nuevos porque soporta tool calls, metadatos de uso de tokens, y datos personalizados dentro del mismo stream — funcionalidades que toTextStreamResponse() no puede transmitir.

Manejo de errores: más allá del try/catch

El error handling que ya tenemos en el pump del controlador cubre los fallos en el stream activo. Pero hay errores que ocurren antes del stream — cuando la API de Anthropic devuelve un 429 (rate limit) o un 500:

// src/ai/ai.controller.ts — versión con manejo de errores completo
import { APICallError } from 'ai';

@Post('chat')
@HttpCode(HttpStatus.OK)
async chat(
  @Body() body: ChatRequestDto,
  @Res() res: Response,
): Promise<void> {
  try {
    const messages = body.messages as CoreMessage[];
    const result = this.aiService.streamChat(messages);
    const streamResponse = result.toUIMessageStreamResponse();

    streamResponse.headers.forEach((value, key) => {
      res.setHeader(key, value);
    });
    res.status(streamResponse.status);

    if (streamResponse.body) {
      const reader = streamResponse.body.getReader();

      const pump = async () => {
        while (true) {
          const { done, value } = await reader.read();
          if (done) { res.end(); break; }
          res.write(value);
        }
      };

      await pump();
    }
  } catch (error) {
    if (APICallError.isInstance(error)) {
      // Error de la API del LLM (429, 500, etc.)
      console.error('[AiController] Error API LLM:', error.message, error.statusCode);

      if (!res.headersSent) {
        const statusCode = error.statusCode === 429 ? 429 : 502;
        res.status(statusCode).json({
          error: error.statusCode === 429
            ? 'Demasiadas peticiones al modelo. Inténtalo en unos segundos.'
            : 'Error al conectar con el modelo de IA.',
        });
      } else {
        res.end();
      }
    } else {
      console.error('[AiController] Error inesperado:', error);
      if (!res.headersSent) {
        res.status(500).json({ error: 'Error interno del servidor.' });
      } else {
        res.end();
      }
    }
  }
}

APICallError.isInstance(error) es el type guard del AI SDK para distinguir errores de la API del LLM de errores genéricos. Útil para devolver mensajes de error específicos al cliente sin exponer detalles internos.

Ejecutar el servidor

# Desarrollo con hot reload
npm run start:dev

# Producción
npm run build && npm run start:prod

El servidor levanta en http://localhost:3000. Prueba el endpoint:

curl -X POST http://localhost:3000/api/chat \
  -H "Content-Type: application/json" \
  -d '{"messages": [{"role": "user", "content": "Qué es NestJS en una frase"}]}' \
  --no-buffer

Verás los chunks SSE llegar en tiempo real en la terminal. Eso confirma que el streaming funciona.

El AiModule en producción: qué añadir después

Lo que hemos construido es una base sólida. En un entorno de producción real, los siguientes pasos son:

1. Autenticación. Añadir un AuthGuard de JWT al endpoint chat para que solo usuarios autenticados consuman tokens. Sin esto, cualquiera con la URL puede vaciar tu cuota.

2. Logging estructurado. Usar @nestjs/winston o Pino para loguear cada request con userId, messageCount, y tokensUsed. El AI SDK expone usage en el stream — puedes capturarlo en el onFinish callback de streamText.

3. Persistencia del historial. El backend actual es stateless — el historial viene del cliente en cada request. En producción con usuarios autenticados, guarda el historial en base de datos y envía solo el conversationId desde el frontend. El servidor reconstruye el historial antes de llamar al modelo.

4. Selección de modelo por request. Si tu app da a los usuarios la opción de elegir entre Claude Sonnet y Claude Haiku (más barato), añade un campo model al DTO y pásalo al servicio. La abstracción del AI SDK hace que el cambio sea trivial.

Si quieres profundizar en este tipo de decisiones de arquitectura — cómo estructurar un producto completo con IA desde la idea hasta producción — en el curso Construye con IA: de la idea al producto con Claude Code lo vemos con proyectos reales, no con demos de laboratorio.

FAQ

¿Puedo usar este módulo con Fastify en lugar de Express?

Sí, pero el pump manual del controlador cambia. Fastify usa Reply en lugar de Response de Express, y el método para escribir chunks es reply.raw.write(). El @Res() res: Response del controlador funcionará si configuras passThrough: true en el decorador: @Res({ passThrough: false }). La lógica del pump en sí no cambia — solo los métodos de la respuesta.

¿El rate limiting con ThrottlerGuard funciona bien detrás de un proxy o load balancer?

Por defecto, ThrottlerGuard usa la IP del request. Si tu app está detrás de un proxy (Nginx, Cloudflare, etc.), la IP será siempre la del proxy. Configura ThrottlerModule con throttlers y usa ThrottlerGuard extendido que lea X-Forwarded-For. Alternativamente, delega el rate limiting al proxy — Nginx tiene limit_req_zone para esto.

¿Cómo evito que el stream consuma tokens si el cliente desconecta?

streamText del AI SDK no cancela automáticamente la petición a Anthropic cuando el cliente cierra la conexión HTTP. Para implementar cancelación, pasa un AbortSignal a streamText:

streamChat(messages: CoreMessage[], signal?: AbortSignal) {
  return streamText({
    model: this.anthropic('claude-sonnet-4-6'),
    messages,
    abortSignal: signal,
  });
}

En el controlador, escucha el evento close de la respuesta y llama a abortController.abort(). Esto cancela la llamada a la API antes de que el modelo termine de generar.

¿Puedo usar @ai-sdk/openai o @ai-sdk/google en lugar de Anthropic?

Sí. Cambia createAnthropic por createOpenAI o createGoogleGenerativeAI en AiService y actualiza el nombre del modelo. El resto del módulo — controlador, DTO, rate limiting, manejo de errores — no cambia. Esa es exactamente la ventaja de usar el AI SDK como capa de abstracción: cambias de proveedor en un sitio.

¿CoreMessage[] es compatible con el formato de mensajes que manda el componente Angular del post anterior?

CoreMessage del AI SDK acepta objetos con role ('user', 'assistant', 'system') y content (string). El ChatMessage del componente Angular del post anterior tiene exactamente esa forma. El cast body.messages as CoreMessage[] funciona directamente — no necesitas transformar nada.

Cierre

Un backend de streaming de IA no es complicado. Lo que sí es complicado es hacerlo bien desde el principio: que valide los inputs, que no queme tokens cuando el cliente desconecta, que no se caiga cuando Anthropic devuelve un 429, que tenga un límite razonable de peticiones por IP.

NestJS más el Vercel AI SDK resuelven ese conjunto de problemas con una arquitectura que ya conoces si llevas tiempo en el ecosistema TypeScript. No hay magia — hay módulos, servicios, inyección de dependencias, y un stream que fluye limpio de principio a fin.

El AiModule que has construido hoy es reutilizable. Impórtalo en cualquier NestJS existente, ajusta el system prompt y el modelo, y tienes un endpoint de IA en producción en menos de una hora.

Si quieres llevarlo más lejos — tool calls, agentes con memoria, pipelines de documentos — en Dominicode Labs tenemos los proyectos completos con los patrones que usamos en producción, incluyendo ejemplos de NestJS con AI SDK con autenticación, persistencia y cancelación de streams.

Por Bezael Pérez — Developer senior con más de 15 años de experiencia y fundador de Dominicode.