SDK Quick Reference

const server = await McpApplicationFactory.create(AppModule);
await server.start();

@McpApp({
  module: AppModule,
  server: { name: 'my-server', version: '1.0.0' },
})
@Module({
  name: 'app',
  imports: [ConfigModule.forRoot(), ProductsModule],
})
export class AppModule {}

@Module({
  name: 'products',
  controllers: [ProductsTools],
  providers: [ProductService],
  exports: [ProductService],
})
export class ProductsModule {}

@Injectable()
export class ProductService {
  async findById(id: string) { /* ... */ }
}

// Ask for it in a constructor
constructor(private products: ProductService) {}

from nitrostack import McpApplicationFactory

app = await McpApplicationFactory.create(AppModule)
await app.start()

@module(
    name="app",
    imports=[ConfigModule.for_root(env_file_path=".env")],
    controllers=[ProductTools],
)
class AppModule: ...

# Optional: name the server
# @mcp_app(module=AppModule, server=ServerConfig(name="my-server", version="1.0.0"))

@module(
    name="products",
    controllers=[ProductTools],
    providers=[ProductService],
    exports=[ProductService],
)
class ProductsModule: ...

@injectable(deps=[Database])  # constructor args must match deps
class ProductService:
    def __init__(self, db: Database):
        self.db = db

import { ToolDecorator as Tool, z } from '@nitrostack/core';

@Tool({
  name: 'get_product',
  description: 'Look up a product by ID',
  inputSchema: z.object({ id: z.string() }),
})
async getProduct(input: { id: string }, ctx: ExecutionContext) {
  return this.products.findById(input.id);
}

import { ResourceDecorator as Resource } from '@nitrostack/core';

@Resource({ uri: 'docs://guide', name: 'Guide', mimeType: 'text/markdown' })
async guide(uri: string, ctx: ExecutionContext) {
  return { contents: [{ uri, mimeType: 'text/markdown', text: '# Hi' }] };
}

import { PromptDecorator as Prompt } from '@nitrostack/core';

@Prompt({ name: 'review_code', description: 'Ask for a code review' })
async review(args: { language: string }, ctx: ExecutionContext) {
  return [{ role: 'user' as const,
    content: { type: 'text' as const, text: `Review this ${args.language} code` } }];
}

@Tool({ name: 'show_product', description: 'Show a product' })
@Widget('product-card') // src/widgets/app/product-card/page.tsx
async showProduct(input: { id: string }, ctx: ExecutionContext) {
  return this.products.findById(input.id);
}

from nitrostack import tool, ExecutionContext

@tool(
    name="get_product",
    description="Look up a product by ID",
    input_schema=GetProductInput,
)
async def get_product(self, input: GetProductInput, context: ExecutionContext) -> dict:
    ...

from nitrostack import resource

@resource(
    uri="docs://guide",
    name="Guide",
    description="The product guide",
    mime_type="text/markdown",
)
async def guide(self, uri: str, context: ExecutionContext) -> dict:
    ...

from nitrostack import prompt, PromptArgument

@prompt(
    name="review_code",
    description="Ask for a code review",
    arguments=[PromptArgument(name="language", description="Programming language", required=True)],
)
async def review(self, args: dict, context: ExecutionContext):
    ...  # return a list of messages

@tool(name="show_product", description="Show a product", input_schema=ShowInput)
@widget("product-card")  # HTML at widgets/out/product-card.html
async def show_product(self, input: ShowInput, context: ExecutionContext) -> dict:
    ...

@Injectable()
export class AdminGuard implements Guard {
  canActivate(ctx: ExecutionContext) {
    return ctx.auth?.scopes?.includes('admin') ?? false;
  }
}

@UseGuards(AdminGuard)

@Middleware()
export class Timing implements MiddlewareInterface {
  async use(ctx: ExecutionContext, next: () => Promise<any>) {
    const start = Date.now();
    const result = await next();
    ctx.logger.info('done', { ms: Date.now() - start });
    return result;
  }
}

@UseMiddleware(Timing)

@Interceptor()
export class Wrap implements InterceptorInterface {
  async intercept(ctx: ExecutionContext, next: () => Promise<any>) {
    return { data: await next() };
  }
}

@UseInterceptors(Wrap)

@Pipe()
export class Trim implements PipeInterface {
  transform(value: any, metadata: ArgumentMetadata) {
    return typeof value === 'string' ? value.trim() : value;
  }
}

@UsePipes(Trim)

@ExceptionFilter()
export class Errors implements ExceptionFilterInterface {
  catch(exception: unknown, ctx: ExecutionContext) {
    return { error: String(exception) };
  }
}

@UseFilters(Errors)

ctx.logger.info('message', { key: 'value' });
ctx.requestId;      // trace a request
ctx.toolName;       // current tool
ctx.auth?.subject;  // set by guards

class AdminGuard:
    async def can_activate(self, context: ExecutionContext) -> bool:
        return "admin" in (context.auth.scopes or [])

@use_guards(AdminGuard)  # built in: ApiKeyGuard, JwtGuard, OAuthGuard

class Timing:
    async def use(self, context: ExecutionContext, next_fn):
        result = await next_fn()
        context.logger.info("done")
        return result

@use_middleware(Timing)

class Wrap:
    async def intercept(self, context: ExecutionContext, next_fn):
        return {"data": await next_fn()}

@use_interceptors(Wrap)

class Trim:
    def transform(self, value, metadata):
        return value.strip() if isinstance(value, str) else value

@use_pipes(Trim)

class Errors:
    def catch(self, exception, context: ExecutionContext):
        return {"error": str(exception)}

@use_filters(Errors)

context.logger.info("message")
context.request_id   # trace a request
context.metadata     # headers, authorization
context.auth         # set by guards

@Cache({ ttl: 600, key: (input) => `search:${input.query}` })

@RateLimit({ requests: 100, window: '1h' })

@HealthCheck({ name: 'database' })
export class DbHealth implements HealthCheckInterface {
  async check() {
    return { status: 'up', message: 'Database connected' };
  }
}

emitEvent('order.created', { orderId });

@OnEvent('order.created')
async onOrder(data: { orderId: string }) { /* ... */ }

@cache(ttl=600)

@rate_limit(max=100, window=3600)  # window in seconds

@health_check(name="database")
async def check_database(self):
    ...

@on_event("order.created")
async def on_order(self, data):
    ...

ApiKeyModule.for_root(...)
JWTModule.for_root(...)
OAuthModule.for_root(...)

from nitrostack import has_scope, has_any_scope, has_all_scopes, require_scopes

# context.auth.aud is a list

@tool(name="export", description="...", input_schema=ExportInput, task_support="optional")
async def export(self, input: ExportInput, context: ExecutionContext) -> dict:
    context.task.update_progress(...)
    ...

from nitrostack.testing import NitroTestingModule

h = await NitroTestingModule.create(AppModule)
await h.call_tool("add", {"a": 5, "b": 10})
await h.read_resource("calc://info")
await h.get_prompt("name", {})