diff --git a/README.md b/README.md index ac53f4729..a8a2336b3 100644 --- a/README.md +++ b/README.md @@ -83,6 +83,7 @@ agentcore # interactive TUI ├── gateway # inspect AgentCore Gateways │ ├── get # get a Gateway by id │ ├── list # list Gateways (server-side paginated) +│ ├── invoke # send a headless request through a Gateway │ ├── target │ │ ├── get # get a Target under a Gateway │ │ └── list # list Targets under a Gateway @@ -159,6 +160,7 @@ agentcore memory record list --memory --namespace --max-r # Inspect Gateway resources without project configuration or deployment agentcore gateway get --id agentcore gateway list --max-results 20 +agentcore gateway invoke --id --payload file://request.json agentcore gateway target get --gateway-id --target-id agentcore gateway target list --gateway-id --max-results 20 agentcore gateway connector get --gateway-id --id @@ -212,6 +214,64 @@ Source-aware values: any field flag documented as such accepts the value inline, `file://` convention). A command reads stdin from at most one flag. For example, `--instructions file://order-quality.txt` or `--instructions -`. +### Invoke a Gateway + +Gateway Invoke is a headless, project-independent HTTP request command. It gets +the Gateway by ID, uses the returned HTTPS origin, selects authentication from +the Gateway's authorizer, and preserves the request and response bodies. + +```bash +# MCP Gateway: use the exact gatewayUrl returned by GetGateway. +agentcore gateway invoke \ + --id \ + --payload '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"agentcore-cli","version":"1"}}}' \ + --accept 'application/json, text/event-stream' \ + --mcp-protocol-version 2025-03-26 + +# HTTP target: --path is relative to the Gateway origin. +agentcore gateway invoke \ + --id \ + --path support-agent/invocations \ + --payload file://request.json \ + --session-id + +# Inference target. +agentcore gateway invoke \ + --id \ + --path inference/v1/messages \ + --payload file://message.json \ + --json + +# GET requests do not accept a payload. +agentcore gateway invoke \ + --id \ + --method GET \ + --path inference/v1/models +``` + +`--path` replaces the path in the returned Gateway URL while retaining its +origin. It must remain relative to the selected Gateway and may include a query +string. Omitting it uses the returned `gatewayUrl` exactly. Supported methods +are `GET`, `POST` (the default), and `DELETE`. POST requires `--payload`; DELETE +may include one. Payloads accept inline bytes, `file://`, or `-` for stdin. + +Authentication follows `GetGateway.authorizerType`: `AWS_IAM` and +`AUTHENTICATE_ONLY` requests use SigV4, `CUSTOM_JWT` requires `--bearer-token`, +and `NONE` uses unsigned HTTPS. Bearer tokens accept inline, `file://`, or stdin +sources; payload and token cannot both read stdin. + +Raw responses stream exact bytes to stdout. `--output-file` streams those bytes +to disk, while `--json` buffers one envelope containing status, selected session +and request metadata, body encoding, and body. Binary or unknown output requires +`--output-file` or `--json` when stdout is a terminal. Response metadata goes to +stderr in raw and file modes. Redirects are returned without being followed. +Non-2xx response bodies use the selected output mode before the command exits +with a failure status. + +Gateway Invoke V1 has no TUI, required request-type selector, tool/model +discovery command, or protocol-specific payload builder. Callers provide the +Gateway-relative route and protocol payload directly. + ### Invoke a Runtime Headless invocation accepts inline, file, or stdin payload bytes: diff --git a/src/core/gateway.test.ts b/src/core/gateway.test.ts index 2322c0c96..1c3f9f2d7 100644 --- a/src/core/gateway.test.ts +++ b/src/core/gateway.test.ts @@ -7,6 +7,7 @@ import { type TargetSummary, } from "@aws-sdk/client-bedrock-agentcore-control"; import { ResultTruncationError } from "../errors"; +import { createSilentLogger } from "../testing"; import type { AwsClients } from "./types"; import { GatewayClient } from "./gateway"; @@ -23,9 +24,13 @@ function ordinary(targetId: string): TargetSummary { function gatewayClient( send: (command: GetGatewayTargetCommand | ListGatewayTargetsCommand) => Promise, ): GatewayClient { - return new GatewayClient({ - control: () => ({ send: mock(send) }) as never, - } as unknown as AwsClients); + return new GatewayClient( + { + control: () => ({ send: mock(send) }) as never, + } as unknown as AwsClients, + globalThis.fetch, + createSilentLogger(), + ); } describe("GatewayClient Connector facade", () => { diff --git a/src/core/gateway.tsx b/src/core/gateway.tsx index b1d7764c5..45cd5b3cc 100644 --- a/src/core/gateway.tsx +++ b/src/core/gateway.tsx @@ -27,15 +27,36 @@ import type { CreateGatewayInput, CreateGatewayRuleInput, CreateGatewayTargetInput, + GatewayInvokeRequest, + GatewayInvokeResponse, } from "../handlers/gateway/types"; -import type { AwsClients, CoreOptions } from "./types"; +import type { Logger } from "../logging"; +import { abortable } from "./abortable"; +import type { AwsClients, CoreFetch, CoreOptions } from "./types"; import { toClientConfig } from "./utils"; const DEFAULT_CONNECTOR_PAGE_SIZE = 100; const MAX_CONNECTOR_TARGET_PAGES = 101; +async function* emptyBody(): AsyncGenerator {} + +function toQuery(url: URL): Record { + const query: Record = {}; + for (const [name, value] of url.searchParams) { + const previous = query[name]; + if (previous === undefined) query[name] = value; + else if (Array.isArray(previous)) previous.push(value); + else query[name] = [previous, value]; + } + return query; +} + export class GatewayClient implements CoreGatewayClient { - constructor(private readonly clients: AwsClients) {} + constructor( + private readonly clients: AwsClients, + private readonly fetch: CoreFetch, + private readonly logger: Logger, + ) {} async createGateway( input: CreateGatewayInput, @@ -52,10 +73,122 @@ export class GatewayClient implements CoreGatewayClient { ); } - async getGateway(id: string, options: CoreOptions): Promise { + async invokeGateway( + request: GatewayInvokeRequest, + options: CoreOptions, + signal?: AbortSignal, + ): Promise { + const logger = this.logger.child({ + operation: "invokeGateway", + authMode: request.authorizerType, + gatewayId: request.gatewayId, + method: request.method, + region: options.region, + }); + const url = new URL(request.url); + if (url.protocol !== "https:") { + throw new TypeError("Gateway invocation requires an HTTPS URL"); + } + + const headers = new Headers(request.applicationHeaders); + try { + if (request.contentType !== undefined) headers.set("Content-Type", request.contentType); + if (request.accept !== undefined) headers.set("Accept", request.accept); + if (request.runtimeSessionId !== undefined) { + headers.set("X-Amzn-Bedrock-AgentCore-Runtime-Session-Id", request.runtimeSessionId); + } + if (request.mcpSessionId !== undefined) { + headers.set("Mcp-Session-Id", request.mcpSessionId); + } + if (request.mcpProtocolVersion !== undefined) { + headers.set("Mcp-Protocol-Version", request.mcpProtocolVersion); + } + if (request.authorizerType === "CUSTOM_JWT") { + headers.set("Authorization", `Bearer ${request.bearerToken}`); + } + } catch { + throw new TypeError("Invalid Gateway request header"); + } + + let fetchHeaders: RequestInit["headers"] = headers; + try { + if (request.authorizerType === "AWS_IAM" || request.authorizerType === "AUTHENTICATE_ONLY") { + const client = this.clients.data(toClientConfig(options)); + const signer = await client.config.signer({ + name: "sigv4", + signingName: "bedrock-agentcore", + signingRegion: options.region, + properties: {}, + }); + const signed = await signer.sign({ + method: request.method, + protocol: url.protocol, + hostname: url.hostname, + ...(url.port && { port: Number(url.port) }), + path: url.pathname, + query: toQuery(url), + headers: { + ...Object.fromEntries(headers.entries()), + host: url.host, + }, + ...(request.payload !== undefined && { body: request.payload }), + }); + fetchHeaders = signed.headers; + } + + const response = await this.fetch(url, { + method: request.method, + redirect: "manual", + headers: fetchHeaders, + ...(request.payload !== undefined && { + body: request.payload as RequestInit["body"], + }), + signal, + }); + if (!response.ok) { + logger + .child({ httpStatusCode: response.status }) + .debug("Gateway invocation returned a non-success response"); + } + + const body = (response.body as AsyncIterable | null) ?? emptyBody(); + return { + statusCode: response.status, + contentType: response.headers.get("content-type") ?? "", + runtimeSessionId: + response.headers.get("x-amzn-bedrock-agentcore-runtime-session-id") ?? undefined, + mcpSessionId: response.headers.get("mcp-session-id") ?? undefined, + mcpProtocolVersion: response.headers.get("mcp-protocol-version") ?? undefined, + requestId: + response.headers.get("x-amzn-requestid") ?? + response.headers.get("x-amz-request-id") ?? + undefined, + body: signal ? abortable(body, signal) : body, + }; + } catch (error) { + if (signal?.aborted) throw signal.reason ?? error; + logger + .child({ + errorName: + error instanceof TypeError + ? "TypeError" + : error instanceof Error + ? "Error" + : typeof error, + }) + .debug("Gateway invocation transport failed"); + throw new Error("Gateway invocation failed"); + } + } + + async getGateway( + id: string, + options: CoreOptions, + signal?: AbortSignal, + ): Promise { return this.clients .control(toClientConfig(options)) - .send(new GetGatewayCommand({ gatewayIdentifier: id })); + .send(new GetGatewayCommand({ gatewayIdentifier: id }), { abortSignal: signal }); } async listGateways( diff --git a/src/core/gatewayInvoke.test.ts b/src/core/gatewayInvoke.test.ts new file mode 100644 index 000000000..c0563eb30 --- /dev/null +++ b/src/core/gatewayInvoke.test.ts @@ -0,0 +1,283 @@ +import { describe, expect, test } from "bun:test"; +import type { BedrockAgentCoreControlClient } from "@aws-sdk/client-bedrock-agentcore-control"; +import type { BedrockAgentCoreClient } from "@aws-sdk/client-bedrock-agentcore"; +import type { CloudWatchLogsClient } from "@aws-sdk/client-cloudwatch-logs"; +import type { IAMClient } from "@aws-sdk/client-iam"; +import type { GatewayInvokeRequest } from "../handlers/gateway/types"; +import { createSilentLogger } from "../testing"; +import { CoreClient } from "./index"; +import type { CoreFetch } from "./types"; + +function request(overrides: Partial = {}): GatewayInvokeRequest { + return { + gatewayId: "gateway-123", + url: "https://gateway.example.test/mcp", + method: "POST", + authorizerType: "NONE", + payload: new TextEncoder().encode("{}"), + contentType: "application/json", + ...overrides, + }; +} + +function coreWithFetch( + fetch: CoreFetch, + sign?: (request: { + method: string; + protocol: string; + hostname: string; + port?: number; + path: string; + query?: Record; + headers: Record; + body?: unknown; + }) => Promise<{ + headers: Record; + }>, +): CoreClient { + return new CoreClient({ + createControlClient: (config) => + ({ config, send: async () => ({}) }) as unknown as BedrockAgentCoreControlClient, + createDataClient: (config) => + ({ + config: { + ...config, + signer: async () => ({ + sign: + sign ?? (async (signedRequest: { headers: Record }) => signedRequest), + }), + }, + }) as unknown as BedrockAgentCoreClient, + createIamClient: (config) => ({ config }) as unknown as IAMClient, + createLogsClient: (config) => ({ config }) as unknown as CloudWatchLogsClient, + fetch, + logger: createSilentLogger(), + }); +} + +describe("Gateway invoke Core transport", () => { + test.each(["CUSTOM_JWT"] as const)( + "sends exact bytes and bearer authentication for %s", + async (authorizerType) => { + const calls: { input: string | URL | Request; init?: RequestInit }[] = []; + const payload = Uint8Array.from([0, 255, 1]); + const core = coreWithFetch(async (input, init) => { + calls.push({ input, init }); + return new Response(Buffer.from("ok"), { + status: 200, + headers: { + "Content-Type": "text/plain", + "Mcp-Session-Id": "returned-mcp", + "Mcp-Protocol-Version": "2025-06-18", + "X-Amzn-RequestId": "request-123", + }, + }); + }); + const controller = new AbortController(); + + const response = await core.gateway.invokeGateway( + request({ + authorizerType, + url: "https://gateway.example.test/target/invocations?trace=true", + payload, + bearerToken: "secret-token", + accept: "text/event-stream", + applicationHeaders: [["X-Tenant", "retail"]], + runtimeSessionId: "runtime-session", + mcpSessionId: "mcp-session", + mcpProtocolVersion: "2025-06-18", + }), + { region: "us-west-2", endpointUrl: "https://control.example.test" }, + controller.signal, + ); + + expect(calls).toHaveLength(1); + expect(String(calls[0]!.input)).toBe( + "https://gateway.example.test/target/invocations?trace=true", + ); + expect(calls[0]!.init).toMatchObject({ + method: "POST", + redirect: "manual", + body: payload, + signal: controller.signal, + }); + expect(new Headers(calls[0]!.init!.headers)).toEqual( + new Headers({ + Accept: "text/event-stream", + Authorization: "Bearer secret-token", + "Content-Type": "application/json", + "Mcp-Protocol-Version": "2025-06-18", + "Mcp-Session-Id": "mcp-session", + "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": "runtime-session", + "X-Tenant": "retail", + }), + ); + expect(response).toMatchObject({ + statusCode: 200, + contentType: "text/plain", + mcpSessionId: "returned-mcp", + mcpProtocolVersion: "2025-06-18", + requestId: "request-123", + }); + }, + ); + + test.each(["AWS_IAM", "AUTHENTICATE_ONLY"] as const)( + "signs %s requests with the data client's resolved signer", + async (authorizerType) => { + const fetchCalls: { input: string | URL | Request; init?: RequestInit }[] = []; + const signedRequests: unknown[] = []; + const core = coreWithFetch( + async (input, init) => { + fetchCalls.push({ input, init }); + return new Response(undefined, { status: 204 }); + }, + async (requestToSign) => { + signedRequests.push(requestToSign); + return { + ...requestToSign, + headers: { + ...requestToSign.headers, + authorization: "AWS4-HMAC-SHA256 signed", + "x-amz-date": "20260810T000000Z", + }, + }; + }, + ); + + await core.gateway.invokeGateway( + request({ + authorizerType, + url: "https://gateway.example.test:8443/path?tag=one&tag=two&space=a+b", + }), + { region: "us-east-1", endpointUrl: "https://control.example.test" }, + ); + + expect(signedRequests).toEqual([ + { + method: "POST", + protocol: "https:", + hostname: "gateway.example.test", + port: 8443, + path: "/path", + query: { tag: ["one", "two"], space: "a b" }, + headers: { + "content-type": "application/json", + host: "gateway.example.test:8443", + }, + body: new TextEncoder().encode("{}"), + }, + ]); + expect(new Headers(fetchCalls[0]!.init!.headers).get("authorization")).toBe( + "AWS4-HMAC-SHA256 signed", + ); + }, + ); + + test("sends NONE requests unsigned and supports GET without a body", async () => { + let init: RequestInit | undefined; + let signerCalled = false; + const core = coreWithFetch( + async (_input, requestInit) => { + init = requestInit; + return new Response(undefined, { status: 204 }); + }, + async (requestToSign) => { + signerCalled = true; + return requestToSign; + }, + ); + + await core.gateway.invokeGateway( + request({ + method: "GET", + authorizerType: "NONE", + payload: undefined, + contentType: undefined, + }), + { region: "us-east-1" }, + ); + + expect(signerCalled).toBe(false); + expect(init).toMatchObject({ method: "GET", redirect: "manual" }); + expect(init).not.toHaveProperty("body"); + expect(new Headers(init!.headers).has("authorization")).toBe(false); + }); + + test.each([ + [302, "redirect response"], + [405, "Method Not Allowed"], + ])("returns HTTP %d bodies without following or discarding them", async (status, content) => { + let init: RequestInit | undefined; + const core = coreWithFetch(async (_input, requestInit) => { + init = requestInit; + return new Response(content, { + status, + headers: { "Content-Type": "text/plain" }, + }); + }); + + const response = await core.gateway.invokeGateway( + request({ + authorizerType: "CUSTOM_JWT", + bearerToken: "secret-token", + applicationHeaders: [["X-Secret", "secret-header"]], + }), + { region: "us-east-1" }, + ); + const chunks: Uint8Array[] = []; + for await (const chunk of response.body) chunks.push(chunk); + + expect(init?.redirect).toBe("manual"); + expect(response.statusCode).toBe(status); + expect(Buffer.concat(chunks).toString()).toBe(content); + }); + + test("sanitizes transport failures", async () => { + const core = coreWithFetch(async () => { + throw new Error("failed with Bearer secret-token"); + }); + + await expect( + core.gateway.invokeGateway( + request({ authorizerType: "CUSTOM_JWT", bearerToken: "secret-token" }), + { region: "us-east-1" }, + ), + ).rejects.toThrow(/^Gateway invocation failed$/); + }); + + test("aborts an established response stream", async () => { + const source = (async function* () { + yield Buffer.from("partial"); + await new Promise(() => {}); + })(); + const core = coreWithFetch(async () => { + return { + ok: true, + status: 200, + headers: new Headers({ "Content-Type": "text/plain" }), + body: source, + } as unknown as Response; + }); + const controller = new AbortController(); + const response = await core.gateway.invokeGateway( + request(), + { region: "us-east-1" }, + controller.signal, + ); + const iterator = response.body[Symbol.asyncIterator](); + + expect(await iterator.next()).toEqual({ done: false, value: Buffer.from("partial") }); + const pending = iterator.next(); + controller.abort(); + + const result = await Promise.race([ + pending.then( + () => "completed", + (error: Error) => error.name, + ), + new Promise((resolve) => setTimeout(() => resolve("timed out"), 25)), + ]); + expect(result).toBe("AbortError"); + }); +}); diff --git a/src/core/index.tsx b/src/core/index.tsx index 2ed5ff93f..34b9b4795 100644 --- a/src/core/index.tsx +++ b/src/core/index.tsx @@ -61,7 +61,7 @@ export class CoreClient implements AwsClients { readonly identity: IdentityClient = new IdentityClient(this); readonly memory: MemoryClient = new MemoryClient(this); readonly runtime: RuntimeClient; - readonly gateway: GatewayClient = new GatewayClient(this); + readonly gateway: GatewayClient; readonly eval: EvalClient; readonly projectManager: ProjectManager; @@ -74,6 +74,7 @@ export class CoreClient implements AwsClients { this.logger = config.logger; const fetch = config.fetch ?? globalThis.fetch; this.runtime = new RuntimeClient(this, fetch, this.logger.child({ module: "runtime" })); + this.gateway = new GatewayClient(this, fetch, this.logger.child({ module: "gateway" })); // EvalClient shares the injected fetch: dataset content is served from a // presigned S3 URL, outside the SDK seam the other operations use. The logger // is used for batch-evaluation result-log diagnostics. diff --git a/src/errors/errors.tsx b/src/errors/errors.tsx index 27c6edc69..1c72fc907 100644 --- a/src/errors/errors.tsx +++ b/src/errors/errors.tsx @@ -153,6 +153,24 @@ export class RuntimeInvokeResponseError extends AgentCoreCLIError { } } +export class GatewayInvokeInterruptedError extends AgentCoreCLIError { + readonly reported: boolean; + + constructor(cause?: unknown, reported = false) { + super("The operation was aborted", { cause, exitCode: 130 }); + this.name = "AbortError"; + this.reported = reported; + } +} + +export class GatewayInvokeResponseError extends AgentCoreCLIError { + readonly reported = true; + + constructor(message: string, cause?: unknown) { + super(message, { cause }); + } +} + /** Remote content could not be fetched, or is not available yet. */ export class NetworkingError extends AgentCoreCLIError { constructor(message: string, options?: AgentCoreCLIErrorOptions) { diff --git a/src/errors/index.tsx b/src/errors/index.tsx index ed44888ab..427c500b2 100644 --- a/src/errors/index.tsx +++ b/src/errors/index.tsx @@ -3,6 +3,8 @@ export { DeserializationError, EmbeddedAssetNotFoundError, FileWriteError, + GatewayInvokeInterruptedError, + GatewayInvokeResponseError, InputValidationError, InvalidEnvironmentError, NestedProjectError, diff --git a/src/handlers/gateway/gateway.test.tsx b/src/handlers/gateway/gateway.test.tsx index e025c526a..fd4441d87 100644 --- a/src/handlers/gateway/gateway.test.tsx +++ b/src/handlers/gateway/gateway.test.tsx @@ -63,6 +63,7 @@ describe("gateway command hierarchy", () => { "create", "get", "list", + "invoke", "target", "connector", "rule", diff --git a/src/handlers/gateway/index.tsx b/src/handlers/gateway/index.tsx index dbbf51c3b..e0e1369f7 100644 --- a/src/handlers/gateway/index.tsx +++ b/src/handlers/gateway/index.tsx @@ -6,6 +6,7 @@ import type { Core } from "../types"; import { createGatewayConnectorHandler } from "./connector"; import { createCreateGatewayHandler } from "./create"; import { createGetGatewayHandler } from "./get"; +import { createInvokeGatewayHandler } from "./invoke"; import { createListGatewaysHandler } from "./list"; import { createGatewayRuleHandler } from "./rule"; import { createGatewayTargetHandler } from "./target"; @@ -18,6 +19,7 @@ export function createGatewayHandler(core: Core, io: AppIO): Router { .handler(createCreateGatewayHandler(core, io)) .handler(createGetGatewayHandler(core)) .handler(createListGatewaysHandler(core)) + .handler(createInvokeGatewayHandler(core, io)) .handler(createGatewayTargetHandler(core, io)) .handler(createGatewayConnectorHandler(core, io)) .handler(createGatewayRuleHandler(core, io)); diff --git a/src/handlers/gateway/invoke/index.tsx b/src/handlers/gateway/invoke/index.tsx new file mode 100644 index 000000000..14febc0a5 --- /dev/null +++ b/src/handlers/gateway/invoke/index.tsx @@ -0,0 +1,112 @@ +import z from "zod"; +import { + GatewayInvokeInterruptedError, + GatewayInvokeResponseError, + InputValidationError, +} from "../../../errors"; +import type { AppIO } from "../../../io"; +import { ExitCode } from "../../../runnable"; +import { createHandler, flag } from "../../../router"; +import { JsonKey } from "../../keys"; +import type { Core } from "../../types"; +import { coreOptsFromCtx } from "../../utils"; +import type { GatewayInvokeMethod } from "../types"; +import { + gatewayIdSchema, + normalizeGatewayInvokeRequest, + parseGatewayInvokeHeaders, + resolveGatewayInvokeSources, +} from "./request"; +import { writeGatewayInvokeResponse } from "./response"; + +export const createInvokeGatewayHandler = (core: Core, io: AppIO) => + createHandler({ + name: "invoke", + description: "invoke an AgentCore Gateway", + flags: [ + flag("id", "the ID of the Gateway", gatewayIdSchema.optional()), + flag( + "path", + "the path relative to the Gateway origin", + z.string().min(1, "requires a nonempty path").optional(), + { sensitive: true }, + ), + flag("method", "the HTTP request method", z.enum(["GET", "POST", "DELETE"]).optional()), + flag("payload", "the inline payload to send", z.string().optional(), { sensitive: true }), + flag("content-type", "the payload content type", z.string().optional()), + flag("accept", "the accepted response content type", z.string().optional()), + flag("header", "an ordered application header", z.array(z.string()).optional(), { + sensitive: true, + }), + flag("bearer-token", "the Gateway bearer token", z.string().optional(), { + sensitive: true, + }), + flag("session-id", "the Runtime target session ID", z.string().optional()), + flag("mcp-session-id", "the MCP session ID", z.string().optional()), + flag("mcp-protocol-version", "the MCP protocol version", z.string().optional()), + flag( + "output-file", + "the response output file", + z.string().min(1, "requires a nonempty path").optional(), + ), + ], + handle: async (ctx, flags) => { + if (flags.id === undefined) { + throw new InputValidationError("required option '--id ' not specified", { + exitCode: ExitCode.USAGE, + }); + } + + const jsonOutput = ctx.require(JsonKey); + if (jsonOutput && flags["output-file"] !== undefined) { + throw new InputValidationError("--json cannot be used with --output-file"); + } + + const controller = new AbortController(); + const interrupt = () => controller.abort(); + process.once("SIGINT", interrupt); + try { + const applicationHeaders = parseGatewayInvokeHeaders(flags.header); + const sources = await resolveGatewayInvokeSources( + { payload: flags.payload, bearerToken: flags["bearer-token"] }, + io.stdin, + controller.signal, + ); + const options = coreOptsFromCtx(ctx); + const gateway = await core.gateway.getGateway(flags.id, options, controller.signal); + const request = normalizeGatewayInvokeRequest(gateway, { + gatewayId: flags.id, + path: flags.path, + method: flags.method as GatewayInvokeMethod | undefined, + payload: sources.payload, + contentType: flags["content-type"], + accept: flags.accept, + applicationHeaders, + bearerToken: sources.bearerToken, + runtimeSessionId: flags["session-id"], + mcpSessionId: flags["mcp-session-id"], + mcpProtocolVersion: flags["mcp-protocol-version"], + }); + const response = await core.gateway.invokeGateway(request, options, controller.signal); + await writeGatewayInvokeResponse(response, { + stdout: io.stdout, + stderr: io.stderr, + outputFile: flags["output-file"], + json: jsonOutput, + signal: controller.signal, + }); + if (response.statusCode < 200 || response.statusCode >= 300) { + throw new GatewayInvokeResponseError(`HTTP ${response.statusCode}`); + } + } catch (error) { + if (controller.signal.aborted && (error as Error)?.name === "AbortError") { + if (error instanceof GatewayInvokeInterruptedError) throw error; + throw new GatewayInvokeInterruptedError(error); + } + throw error; + } finally { + controller.abort(); + process.off("SIGINT", interrupt); + } + }, + }); diff --git a/src/handlers/gateway/invoke/invoke.test.tsx b/src/handlers/gateway/invoke/invoke.test.tsx new file mode 100644 index 000000000..07ed590f5 --- /dev/null +++ b/src/handlers/gateway/invoke/invoke.test.tsx @@ -0,0 +1,430 @@ +import { describe, expect, test } from "bun:test"; +import { mkdtemp, readFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { PassThrough } from "node:stream"; +import type { GetGatewayResponse } from "@aws-sdk/client-bedrock-agentcore-control"; +import type { AppIO } from "../../../io"; +import { ExitCode, runWithExitCode } from "../../../runnable"; +import { + createSilentLogger, + TestCoreClient, + TestGlobalConfigAccessor, + waitFor, +} from "../../../testing"; +import { createRootHandler } from "../../index"; +import type { GatewayInvokeRequest } from "../types"; + +const REGION = "us-west-2"; +const GATEWAY_ID = "gateway-123"; +const GATEWAY_URL = "https://gateway-123.gateway.example.test/mcp"; + +function body(...chunks: Uint8Array[]): AsyncIterable { + return (async function* () { + yield* chunks; + })(); +} + +function captureIO(input?: Uint8Array) { + const stdin = new PassThrough() as unknown as NodeJS.ReadStream; + const stdout = new PassThrough(); + const stderr = new PassThrough(); + const stdoutChunks: Buffer[] = []; + const stderrChunks: Buffer[] = []; + stdout.on("data", (chunk) => stdoutChunks.push(Buffer.from(chunk))); + stderr.on("data", (chunk) => stderrChunks.push(Buffer.from(chunk))); + if (input !== undefined) stdin.end(input); + return { + io: { + stdin, + stdout: stdout as unknown as NodeJS.WriteStream, + stderr: stderr as unknown as NodeJS.WriteStream, + } satisfies AppIO, + stdout: () => Buffer.concat(stdoutChunks), + stderr: () => Buffer.concat(stderrChunks), + }; +} + +async function runCommand(core: TestCoreClient, io: AppIO, args: string[]): Promise { + const root = createRootHandler(core, { + io, + logger: createSilentLogger(), + globalConfigAccessor: new TestGlobalConfigAccessor(), + }); + await root.route(["node", "agentcore", ...args, "--region", REGION]); +} + +function configuredCore(gateway: Partial = {}): TestCoreClient { + const core = new TestCoreClient(); + core.gateway + .setGetResponse({ + gatewayId: GATEWAY_ID, + gatewayUrl: GATEWAY_URL, + authorizerType: "NONE", + ...gateway, + } as GetGatewayResponse) + .setInvokeResponse({ + statusCode: 200, + contentType: "application/json", + mcpSessionId: "returned-session", + body: body(Buffer.from([0, 255]), Buffer.from([10, 1])), + }); + return core; +} + +describe("gateway invoke", () => { + test("resolves the Gateway, invokes it, and streams exact response bytes", async () => { + const core = configuredCore(); + const output = captureIO(); + + await runCommand(core, output.io, [ + "gateway", + "invoke", + "--id", + GATEWAY_ID, + "--payload", + '{"jsonrpc":"2.0","id":1,"method":"tools/list"}', + ]); + + expect(core.gateway.calls.map((call) => call.method)).toEqual(["getGateway", "invokeGateway"]); + const lookup = core.gateway.calls[0]!; + const invoke = core.gateway.calls[1]!; + expect(lookup.args.slice(0, 2)).toEqual([GATEWAY_ID, { region: REGION }]); + expect(lookup.args[2]).toBe(invoke.args[2]); + expect(invoke.args[1]).toEqual({ region: REGION }); + expect(invoke.args[0]).toEqual({ + gatewayId: GATEWAY_ID, + url: GATEWAY_URL, + method: "POST", + authorizerType: "NONE", + payload: new TextEncoder().encode('{"jsonrpc":"2.0","id":1,"method":"tools/list"}'), + contentType: "application/json", + }); + expect(output.stdout()).toEqual(Buffer.from([0, 255, 10, 1])); + expect(output.stderr().toString()).toContain("mcp-session-id=returned-session"); + }); + + test("passes method, root-relative path, headers, sessions, and endpoint options", async () => { + const core = configuredCore({ authorizerType: "CUSTOM_JWT" }); + const output = captureIO(); + + await runCommand(core, output.io, [ + "gateway", + "invoke", + "--id", + GATEWAY_ID, + "--path", + "target/invocations?trace=true", + "--method", + "DELETE", + "--payload", + "{}", + "--content-type", + "application/problem+json", + "--accept", + "text/event-stream", + "--header", + "X-Tenant: retail", + "--bearer-token", + "secret-token", + "--session-id", + "runtime-session", + "--mcp-session-id", + "mcp-session", + "--mcp-protocol-version", + "2025-06-18", + "--endpoint-url", + "https://control.example.test", + ]); + + const request = core.gateway.calls.find((call) => call.method === "invokeGateway")! + .args[0] as GatewayInvokeRequest; + expect(request).toMatchObject({ + url: "https://gateway-123.gateway.example.test/target/invocations?trace=true", + method: "DELETE", + authorizerType: "CUSTOM_JWT", + contentType: "application/problem+json", + accept: "text/event-stream", + applicationHeaders: [["X-Tenant", "retail"]], + bearerToken: "secret-token", + runtimeSessionId: "runtime-session", + mcpSessionId: "mcp-session", + mcpProtocolVersion: "2025-06-18", + }); + expect(core.gateway.calls[0]!.args[1]).toEqual({ + region: REGION, + endpointUrl: "https://control.example.test", + }); + }); + + test("supports GET without a payload", async () => { + const core = configuredCore({ gatewayUrl: "https://gateway.example.test" }); + const output = captureIO(); + + await runCommand(core, output.io, [ + "gateway", + "invoke", + "--id", + GATEWAY_ID, + "--method", + "GET", + "--path", + "inference/v1/models", + ]); + + expect( + core.gateway.calls.find((call) => call.method === "invokeGateway")!.args[0], + ).toMatchObject({ + method: "GET", + url: "https://gateway.example.test/inference/v1/models", + }); + }); + + test("passes an explicitly empty payload as zero bytes", async () => { + const core = configuredCore(); + const output = captureIO(); + + await runCommand(core, output.io, ["gateway", "invoke", "--id", GATEWAY_ID, "--payload", ""]); + + const request = core.gateway.calls.find((call) => call.method === "invokeGateway")! + .args[0] as GatewayInvokeRequest; + expect(request.payload).toEqual(new Uint8Array()); + }); + + test("resolves a bearer token from stdin through the command flow", async () => { + const core = configuredCore({ authorizerType: "CUSTOM_JWT" }); + const output = captureIO(Buffer.from("secret-token")); + + await runCommand(core, output.io, [ + "gateway", + "invoke", + "--id", + GATEWAY_ID, + "--payload", + "{}", + "--bearer-token", + "-", + ]); + + const request = core.gateway.calls.find((call) => call.method === "invokeGateway")! + .args[0] as GatewayInvokeRequest; + expect(request.bearerToken).toBe("secret-token"); + }); + + test("buffers the response in JSON mode", async () => { + const core = configuredCore(); + core.gateway.setInvokeResponse({ + statusCode: 200, + contentType: "application/json", + requestId: "request-123", + body: body(Buffer.from('{"ok":true}')), + }); + const output = captureIO(); + + await runCommand(core, output.io, [ + "gateway", + "invoke", + "--id", + GATEWAY_ID, + "--payload", + "{}", + "--json", + ]); + + expect(JSON.parse(output.stdout().toString())).toMatchObject({ + statusCode: 200, + requestId: "request-123", + bodyEncoding: "utf8", + body: '{"ok":true}', + complete: true, + }); + expect(output.stderr()).toHaveLength(0); + }); + + test("streams a non-2xx body before returning a failure exit code", async () => { + const core = configuredCore(); + core.gateway.setInvokeResponse({ + statusCode: 405, + contentType: "text/plain", + requestId: "request-405", + body: body(Buffer.from("Method Not Allowed")), + }); + const output = captureIO(); + + const code = await runWithExitCode(async () => + runCommand(core, output.io, ["gateway", "invoke", "--id", GATEWAY_ID, "--method", "GET"]), + ); + + expect(code).toBe(ExitCode.FAILURE); + expect(output.stdout().toString()).toBe("Method Not Allowed"); + expect(output.stderr().toString()).toContain( + "status=405 content-type=text/plain runtime-session-id=- mcp-session-id=- " + + "mcp-protocol-version=- request-id=request-405 complete=true bytes=18", + ); + }); + + test("writes a non-2xx JSON envelope before returning a failure exit code", async () => { + const core = configuredCore(); + core.gateway.setInvokeResponse({ + statusCode: 422, + contentType: "application/problem+json", + requestId: "request-422", + body: body(Buffer.from('{"message":"invalid request"}')), + }); + const output = captureIO(); + + const code = await runWithExitCode(async () => + runCommand(core, output.io, [ + "gateway", + "invoke", + "--id", + GATEWAY_ID, + "--payload", + "{}", + "--json", + ]), + ); + + expect(code).toBe(ExitCode.FAILURE); + expect(JSON.parse(output.stdout().toString())).toEqual({ + statusCode: 422, + contentType: "application/problem+json", + requestId: "request-422", + bodyEncoding: "utf8", + body: '{"message":"invalid request"}', + complete: true, + }); + expect(output.stderr()).toHaveLength(0); + }); + + test("writes a non-2xx body to a file before returning a failure exit code", async () => { + const directory = await mkdtemp(join(tmpdir(), "gateway-invoke-")); + const outputPath = join(directory, "response.bin"); + try { + const core = configuredCore(); + const responseBytes = Buffer.from([0, 255, 10, 1]); + core.gateway.setInvokeResponse({ + statusCode: 503, + contentType: "application/octet-stream", + requestId: "request-503", + body: body(responseBytes), + }); + const output = captureIO(); + + const code = await runWithExitCode(async () => + runCommand(core, output.io, [ + "gateway", + "invoke", + "--id", + GATEWAY_ID, + "--payload", + "{}", + "--output-file", + outputPath, + ]), + ); + + expect(code).toBe(ExitCode.FAILURE); + expect(await readFile(outputPath)).toEqual(responseBytes); + expect(output.stdout()).toHaveLength(0); + expect(output.stderr().toString()).toContain( + "status=503 content-type=application/octet-stream runtime-session-id=- " + + "mcp-session-id=- mcp-protocol-version=- request-id=request-503 " + + "complete=true bytes=4", + ); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + + test.each([ + [["gateway", "invoke", "--payload", "{}"], /--id/], + [["gateway", "invoke", "--id", GATEWAY_ID], /--payload/], + [ + [ + "gateway", + "invoke", + "--id", + GATEWAY_ID, + "--payload", + "{}", + "--json", + "--output-file", + "response.bin", + ], + /--json cannot be used/, + ], + [ + ["gateway", "invoke", "--id", GATEWAY_ID, "--payload", "{}", "--request-type", "mcp"], + /unknown option '--request-type'/, + ], + ] as const)("rejects invalid input before invocation", async (args, message) => { + const core = configuredCore(); + const output = captureIO(); + await expect(runCommand(core, output.io, [...args])).rejects.toThrow(message); + expect(core.gateway.calls.some((call) => call.method === "invokeGateway")).toBe(false); + }); + + test("classifies a missing ID as usage", async () => { + const core = configuredCore(); + const output = captureIO(); + + const code = await runWithExitCode(async () => + runCommand(core, output.io, ["gateway", "invoke", "--payload", "{}"]), + ); + + expect(code).toBe(ExitCode.USAGE); + expect(core.gateway.calls).toEqual([]); + }); + + test("SIGINT aborts lookup and invocation through the same signal", async () => { + const core = configuredCore(); + const output = captureIO(); + core.gateway.invokeGateway = async (request, options, signal) => { + core.gateway.calls.push({ method: "invokeGateway", args: [request, options, signal] }); + return new Promise((_, reject) => { + const abort = () => + reject(Object.assign(new Error("transport aborted"), { name: "AbortError" })); + if (signal?.aborted) abort(); + else signal?.addEventListener("abort", abort, { once: true }); + }); + }; + const pending = runCommand(core, output.io, [ + "gateway", + "invoke", + "--id", + GATEWAY_ID, + "--payload", + "{}", + ]); + + try { + await waitFor(() => core.gateway.calls.some((call) => call.method === "invokeGateway")); + process.emit("SIGINT", "SIGINT"); + + await expect(pending).rejects.toMatchObject({ name: "AbortError", reported: false }); + const lookupSignal = core.gateway.calls[0]!.args[2] as AbortSignal; + const invokeSignal = core.gateway.calls[1]!.args[2] as AbortSignal; + expect(lookupSignal).toBe(invokeSignal); + expect(invokeSignal.aborted).toBe(true); + } finally { + await pending.catch(() => undefined); + } + }); + + test("registers a headless invoke leaf without a request-type flag", () => { + const core = configuredCore(); + const output = captureIO(); + const root = createRootHandler(core, { + io: output.io, + logger: createSilentLogger(), + globalConfigAccessor: new TestGlobalConfigAccessor(), + }); + const gateway = root.children().find((child) => child.name() === "gateway"); + const invoke = gateway?.children().find((child) => child.name() === "invoke"); + + expect(invoke).toBeDefined(); + expect(invoke?.flags().map((registered) => registered.name)).not.toContain("request-type"); + expect(invoke?.flags().find((registered) => registered.name === "path")?.sensitive).toBe(true); + }); +}); diff --git a/src/handlers/gateway/invoke/request.test.ts b/src/handlers/gateway/invoke/request.test.ts new file mode 100644 index 000000000..6970200d2 --- /dev/null +++ b/src/handlers/gateway/invoke/request.test.ts @@ -0,0 +1,196 @@ +import { describe, expect, test } from "bun:test"; +import { PassThrough } from "node:stream"; +import type { GetGatewayResponse } from "@aws-sdk/client-bedrock-agentcore-control"; +import { + normalizeGatewayInvokeRequest, + parseGatewayInvokeHeaders, + resolveGatewayInvokeSources, +} from "./request"; + +const GATEWAY_ID = "gateway-123"; + +function stdin(bytes?: Uint8Array): NodeJS.ReadStream { + const stream = new PassThrough(); + if (bytes !== undefined) stream.end(bytes); + return stream as unknown as NodeJS.ReadStream; +} + +function detail(overrides: Partial = {}): GetGatewayResponse { + return { + gatewayUrl: "https://gateway-123.gateway.example.test/mcp", + authorizerType: "NONE", + ...overrides, + } as GetGatewayResponse; +} + +function input(overrides: Record = {}) { + return { + gatewayId: GATEWAY_ID, + payload: new TextEncoder().encode("{}"), + ...overrides, + }; +} + +describe("Gateway invoke sources", () => { + test("resolves payload bytes and bearer-token text through one resolver", async () => { + const payload = Uint8Array.from([0, 255, 10]); + + const result = await resolveGatewayInvokeSources( + { payload: "-", bearerToken: "secret-token" }, + stdin(payload), + ); + + expect(result.payload).toEqual(payload); + expect(result.bearerToken).toBe("secret-token"); + }); + + test("preserves an explicitly empty inline payload", async () => { + const result = await resolveGatewayInvokeSources({ payload: "" }, stdin()); + expect(result.payload).toEqual(new Uint8Array()); + }); + + test("rejects payload and bearer token both reading stdin before consuming it", async () => { + await expect( + resolveGatewayInvokeSources({ payload: "-", bearerToken: "-" }, stdin(Buffer.from("value"))), + ).rejects.toThrow("cannot both read from stdin"); + }); + + test("rejects a bearer token that is not valid UTF-8", async () => { + await expect( + resolveGatewayInvokeSources( + { payload: "{}", bearerToken: "-" }, + stdin(Uint8Array.from([0xff])), + ), + ).rejects.toThrow("must contain valid UTF-8"); + }); +}); + +describe("Gateway invoke headers", () => { + test("parses ordered header values containing additional colons", () => { + expect(parseGatewayInvokeHeaders(["X-One: 1", "X-Url: https://example.test/a:b"])).toEqual([ + ["X-One", "1"], + ["X-Url", "https://example.test/a:b"], + ]); + }); + + test.each([ + [["not-a-header"], "Name: value"], + [["Bad Header: value"], "Invalid HTTP header name"], + [["X-One: 1", "x-one: 2"], "Duplicate header"], + [["Authorization: secret"], "reserved"], + [["Mcp-Session-Id: session"], "reserved"], + [["X-Amz-Date: date"], "reserved"], + ])("rejects invalid header input %j", (headers, message) => { + expect(() => parseGatewayInvokeHeaders(headers)).toThrow(message); + }); +}); + +describe("normalizeGatewayInvokeRequest", () => { + test("uses the exact Gateway URL and defaults POST JSON requests", () => { + expect(normalizeGatewayInvokeRequest(detail(), input())).toEqual({ + gatewayId: GATEWAY_ID, + url: "https://gateway-123.gateway.example.test/mcp", + method: "POST", + authorizerType: "NONE", + payload: new TextEncoder().encode("{}"), + contentType: "application/json", + }); + }); + + test("replaces the returned path from the Gateway origin and preserves query values", () => { + const request = normalizeGatewayInvokeRequest( + detail(), + input({ path: "/inference/v1/messages?stream=true&tag=one&tag=two" }), + ); + + expect(request.url).toBe( + "https://gateway-123.gateway.example.test/inference/v1/messages?stream=true&tag=one&tag=two", + ); + }); + + test("supports GET without a payload and no implicit content type", () => { + expect( + normalizeGatewayInvokeRequest( + detail({ gatewayUrl: "https://gateway.example.test" }), + input({ method: "GET", path: "inference/v1/models", payload: undefined }), + ), + ).toEqual({ + gatewayId: GATEWAY_ID, + url: "https://gateway.example.test/inference/v1/models", + method: "GET", + authorizerType: "NONE", + }); + }); + + test("maps every optional request field", () => { + const request = normalizeGatewayInvokeRequest( + detail({ + authorizerType: "CUSTOM_JWT", + gatewayUrl: "https://gateway.example.test", + }), + input({ + path: "target/invocations", + method: "DELETE", + contentType: "application/problem+json", + accept: "text/event-stream", + applicationHeaders: [["X-Tenant", "retail"]], + bearerToken: "secret", + runtimeSessionId: "runtime-session", + mcpSessionId: "mcp-session", + mcpProtocolVersion: "2025-06-18", + }), + ); + + expect(request).toMatchObject({ + method: "DELETE", + authorizerType: "CUSTOM_JWT", + contentType: "application/problem+json", + accept: "text/event-stream", + applicationHeaders: [["X-Tenant", "retail"]], + bearerToken: "secret", + runtimeSessionId: "runtime-session", + mcpSessionId: "mcp-session", + mcpProtocolVersion: "2025-06-18", + }); + }); + + test.each([ + ["missing URL", detail({ gatewayUrl: undefined }), input(), "returned no invocation URL"], + ["invalid URL", detail({ gatewayUrl: "not a URL" }), input(), "invalid invocation URL"], + [ + "insecure URL", + detail({ gatewayUrl: "http://gateway.example.test" }), + input(), + "requires an HTTPS URL", + ], + ["absolute path", detail(), input({ path: "https://evil.example" }), "must be relative"], + ["network path", detail(), input({ path: "//evil.example/a" }), "must be relative"], + ["fragment", detail(), input({ path: "target/path#fragment" }), "stay within"], + ["dot segment", detail(), input({ path: "target/../secret" }), "cannot contain"], + ["missing POST payload", detail(), input({ payload: undefined }), "--payload"], + ["GET payload", detail(), input({ method: "GET" }), "do not accept --payload"], + [ + "unsupported authorizer", + detail({ authorizerType: undefined }), + input(), + "unsupported authorizer", + ], + ])("rejects %s", (_name, gateway, requestInput, message) => { + expect(() => normalizeGatewayInvokeRequest(gateway, requestInput)).toThrow(message); + }); + + test.each(["CUSTOM_JWT"] as const)("requires a bearer token for %s", (authorizerType) => { + expect(() => normalizeGatewayInvokeRequest(detail({ authorizerType }), input())).toThrow( + "requires --bearer-token", + ); + }); + + test.each(["AUTHENTICATE_ONLY", "AWS_IAM", "NONE"] as const)( + "rejects a bearer token for %s", + (authorizerType) => { + expect(() => + normalizeGatewayInvokeRequest(detail({ authorizerType }), input({ bearerToken: "secret" })), + ).toThrow("does not accept --bearer-token"); + }, + ); +}); diff --git a/src/handlers/gateway/invoke/request.ts b/src/handlers/gateway/invoke/request.ts new file mode 100644 index 000000000..7a4a2a6c1 --- /dev/null +++ b/src/handlers/gateway/invoke/request.ts @@ -0,0 +1,196 @@ +import { validateHeaderName, validateHeaderValue } from "node:http"; +import type { AuthorizerType, GetGatewayResponse } from "@aws-sdk/client-bedrock-agentcore-control"; +import z from "zod"; +import { InputValidationError, SourceResolutionError } from "../../../errors"; +import { SourceResolver } from "../../../io"; +import type { GatewayInvokeMethod, GatewayInvokeRequest } from "../types"; + +export const gatewayIdSchema = z + .string() + .refine((value) => !value.startsWith("arn:"), "must be a Gateway ID, not an ARN"); + +type GatewayInvokeInput = { + gatewayId: string; + path?: string; + method?: GatewayInvokeMethod; + payload?: Uint8Array; + contentType?: string; + accept?: string; + applicationHeaders?: [string, string][]; + bearerToken?: string; + runtimeSessionId?: string; + mcpSessionId?: string; + mcpProtocolVersion?: string; +}; + +const RESERVED_HEADERS = new Set([ + "accept", + "authorization", + "content-length", + "content-type", + "host", + "mcp-protocol-version", + "mcp-session-id", + "x-amz-content-sha256", + "x-amz-date", + "x-amz-security-token", + "x-amzn-bedrock-agentcore-runtime-session-id", +]); + +export async function resolveGatewayInvokeSources( + sources: { payload?: string; bearerToken?: string }, + stdin: NodeJS.ReadStream, + signal?: AbortSignal, +): Promise<{ payload?: Uint8Array; bearerToken?: string }> { + if (sources.payload === "-" && sources.bearerToken === "-") { + throw new InputValidationError("Payload and bearer token cannot both read from stdin"); + } + + const resolver = new SourceResolver({ stdin, signal }); + try { + const payload = await resolver.resolveBytes("payload", sources.payload); + const bearerToken = await resolver.resolveText("bearer-token", sources.bearerToken); + return { + ...(payload !== undefined && { payload }), + ...(bearerToken !== undefined && { bearerToken }), + }; + } catch (error) { + if (error instanceof SourceResolutionError) { + throw new InputValidationError(error.message, { cause: error }); + } + throw error; + } +} + +export function parseGatewayInvokeHeaders(values: string[] = []): [string, string][] { + const seen = new Set(); + + return values.map((header) => { + const separator = header.indexOf(":"); + if (separator < 1) throw new InputValidationError("Header must use 'Name: value' format"); + const name = header.slice(0, separator).trim(); + const value = header.slice(separator + 1).trim(); + try { + validateHeaderName(name); + } catch { + throw new InputValidationError( + `Invalid HTTP header name: ${name} (must use valid HTTP token characters)`, + ); + } + try { + validateHeaderValue(name, value); + } catch { + throw new InputValidationError( + `Invalid header value for ${name}: contains a character not allowed in HTTP headers`, + ); + } + const lower = name.toLowerCase(); + if (seen.has(lower)) throw new InputValidationError(`Duplicate header: ${name}`); + seen.add(lower); + if (RESERVED_HEADERS.has(lower)) { + throw new InputValidationError(`Application header is reserved: ${name}`); + } + return [name, value]; + }); +} + +function resolveGatewayInvokeUrl(gatewayUrl: string | undefined, path?: string): string { + if (!gatewayUrl) throw new InputValidationError("Gateway returned no invocation URL"); + + let base: URL; + try { + base = new URL(gatewayUrl); + } catch (error) { + throw new InputValidationError("Gateway returned an invalid invocation URL", { cause: error }); + } + if (base.protocol !== "https:") { + throw new InputValidationError("Gateway invocation requires an HTTPS URL"); + } + if (base.username || base.password || base.hash) { + throw new InputValidationError("Gateway returned an invalid invocation URL"); + } + if (path === undefined) return base.href; + if (/^[a-zA-Z][a-zA-Z\d+.-]*:/.test(path) || path.startsWith("//")) { + throw new InputValidationError("--path must be relative to the Gateway"); + } + + const relative = path.replace(/^\/+/, ""); + const pathOnly = relative.split(/[?#]/, 1)[0]!; + if (pathOnly.split("/").some((segment) => segment === "." || segment === "..")) { + throw new InputValidationError("--path cannot contain '.' or '..' segments"); + } + + const url = new URL(relative, `${base.origin}/`); + if (url.origin !== base.origin || url.username || url.password || url.hash) { + throw new InputValidationError("--path must stay within the selected Gateway"); + } + return url.href; +} + +function validateAuthorizer( + authorizerType: AuthorizerType | undefined, + bearerToken?: string, +): AuthorizerType { + switch (authorizerType) { + case "CUSTOM_JWT": + if (!bearerToken) { + throw new InputValidationError(`${authorizerType} Gateway requires --bearer-token`); + } + return authorizerType; + case "AUTHENTICATE_ONLY": + case "AWS_IAM": + case "NONE": + if (bearerToken !== undefined) { + throw new InputValidationError(`${authorizerType} Gateway does not accept --bearer-token`); + } + return authorizerType; + default: + throw new InputValidationError("Gateway uses an unsupported authorizer"); + } +} + +export function normalizeGatewayInvokeRequest( + detail: GetGatewayResponse, + input: GatewayInvokeInput, +): GatewayInvokeRequest { + const method = input.method ?? "POST"; + if (method === "POST" && input.payload === undefined) { + throw new InputValidationError("required option '--payload ' not specified"); + } + if (method === "GET" && input.payload !== undefined) { + throw new InputValidationError("GET requests do not accept --payload"); + } + + const authorizerType = validateAuthorizer(detail.authorizerType, input.bearerToken); + const { + gatewayId, + path, + payload, + contentType, + applicationHeaders = [], + accept, + bearerToken, + runtimeSessionId, + mcpSessionId, + mcpProtocolVersion, + } = input; + + return { + gatewayId, + url: resolveGatewayInvokeUrl(detail.gatewayUrl, path), + method, + authorizerType, + ...(payload !== undefined && { payload }), + ...(contentType !== undefined + ? { contentType } + : payload !== undefined + ? { contentType: "application/json" } + : {}), + ...(applicationHeaders.length > 0 && { applicationHeaders }), + ...(accept !== undefined && { accept }), + ...(bearerToken !== undefined && { bearerToken }), + ...(runtimeSessionId !== undefined && { runtimeSessionId }), + ...(mcpSessionId !== undefined && { mcpSessionId }), + ...(mcpProtocolVersion !== undefined && { mcpProtocolVersion }), + }; +} diff --git a/src/handlers/gateway/invoke/response.test.ts b/src/handlers/gateway/invoke/response.test.ts new file mode 100644 index 000000000..ca0b6e831 --- /dev/null +++ b/src/handlers/gateway/invoke/response.test.ts @@ -0,0 +1,105 @@ +import { describe, expect, test } from "bun:test"; +import { PassThrough } from "node:stream"; +import type { GatewayInvokeResponse } from "../types"; +import { writeGatewayInvokeResponse } from "./response"; + +function body(...chunks: Uint8Array[]): AsyncIterable { + return (async function* () { + yield* chunks; + })(); +} + +function capture() { + const stream = new PassThrough(); + const chunks: Buffer[] = []; + stream.on("data", (chunk) => chunks.push(Buffer.from(chunk))); + return { + stream: stream as unknown as NodeJS.WriteStream, + bytes: () => Buffer.concat(chunks), + }; +} + +function response(overrides: Partial = {}): GatewayInvokeResponse { + return { + statusCode: 200, + contentType: "application/json", + body: body(Buffer.from("{}")), + ...overrides, + }; +} + +describe("Gateway invoke response output", () => { + test("streams exact bytes and reports Gateway metadata", async () => { + const stdout = capture(); + const stderr = capture(); + + await writeGatewayInvokeResponse( + response({ + statusCode: 206, + contentType: "text/event-stream", + runtimeSessionId: "runtime-session", + mcpSessionId: "mcp-session", + mcpProtocolVersion: "2025-06-18", + requestId: "request-123", + body: body(Buffer.from([0, 255]), Buffer.from([10, 1])), + }), + { stdout: stdout.stream, stderr: stderr.stream }, + ); + + expect(stdout.bytes()).toEqual(Buffer.from([0, 255, 10, 1])); + expect(stderr.bytes().toString()).toBe( + "status=206 content-type=text/event-stream runtime-session-id=runtime-session " + + "mcp-session-id=mcp-session mcp-protocol-version=2025-06-18 " + + "request-id=request-123 complete=true bytes=4\n", + ); + }); + + test("JSON mode emits one metadata and body envelope", async () => { + const stdout = capture(); + const stderr = capture(); + + await writeGatewayInvokeResponse( + response({ + mcpSessionId: "mcp-session", + requestId: "request-123", + body: body(Buffer.from('{"ok":true}')), + }), + { stdout: stdout.stream, stderr: stderr.stream, json: true }, + ); + + expect(JSON.parse(stdout.bytes().toString())).toEqual({ + statusCode: 200, + contentType: "application/json", + mcpSessionId: "mcp-session", + requestId: "request-123", + bodyEncoding: "utf8", + body: '{"ok":true}', + complete: true, + }); + expect(stderr.bytes()).toHaveLength(0); + }); + + test("preserves partial output and reports a sanitized stream failure", async () => { + const stdout = capture(); + const stderr = capture(); + const upstream = new Error("secret upstream response"); + + await expect( + writeGatewayInvokeResponse( + response({ + body: (async function* () { + yield Buffer.from("partial"); + throw upstream; + })(), + }), + { stdout: stdout.stream, stderr: stderr.stream }, + ), + ).rejects.toMatchObject({ message: "response stream failed", reported: true }); + + expect(stdout.bytes().toString()).toBe("partial"); + expect(stderr.bytes().toString()).toContain( + "complete=false bytes=7 error=response-stream-failed", + ); + expect(stderr.bytes().toString()).not.toContain(upstream.message); + }); +}); diff --git a/src/handlers/gateway/invoke/response.ts b/src/handlers/gateway/invoke/response.ts new file mode 100644 index 000000000..46672f9a1 --- /dev/null +++ b/src/handlers/gateway/invoke/response.ts @@ -0,0 +1,40 @@ +import { GatewayInvokeInterruptedError, GatewayInvokeResponseError } from "../../../errors"; +import { writeStreamingResponse, type StreamingResponseOutput } from "../../../io"; +import type { GatewayInvokeResponse } from "../types"; + +const RESPONSE_STREAM_FAILED = "response stream failed"; + +function failure(error: unknown): never { + const interrupted = (error as Error)?.name === "AbortError"; + if (interrupted) throw new GatewayInvokeInterruptedError(error, true); + throw new GatewayInvokeResponseError(RESPONSE_STREAM_FAILED, error); +} + +function summary( + response: GatewayInvokeResponse, + byteCount: number, + complete: boolean, + error?: string, +): string { + const value = (item?: string) => item || "-"; + return ( + `status=${response.statusCode} content-type=${value(response.contentType)} ` + + `runtime-session-id=${value(response.runtimeSessionId)} ` + + `mcp-session-id=${value(response.mcpSessionId)} ` + + `mcp-protocol-version=${value(response.mcpProtocolVersion)} ` + + `request-id=${value(response.requestId)} ` + + `complete=${complete} bytes=${byteCount}${error ? ` error=${error}` : ""}\n` + ); +} + +export async function writeGatewayInvokeResponse( + response: GatewayInvokeResponse, + output: StreamingResponseOutput, +): Promise { + await writeStreamingResponse(response, output, { + metadata: ({ body: _body, ...metadata }) => metadata, + summary, + fail: failure, + binaryTtyError: "Binary or unknown response content requires --output-file or --json", + }); +} diff --git a/src/handlers/gateway/types.tsx b/src/handlers/gateway/types.tsx index c414d1c3b..3ca50577e 100644 --- a/src/handlers/gateway/types.tsx +++ b/src/handlers/gateway/types.tsx @@ -1,4 +1,5 @@ import type { + AuthorizerType, CreateGatewayRequest, CreateGatewayResponse, CreateGatewayRuleRequest, @@ -24,9 +25,41 @@ export type CreateGatewayTargetInput = CreateGatewayTargetRequest; export type CreateGatewayRuleInput = CreateGatewayRuleRequest; +export type GatewayInvokeMethod = "GET" | "POST" | "DELETE"; + +export type GatewayInvokeRequest = { + gatewayId: string; + url: string; + method: GatewayInvokeMethod; + authorizerType: AuthorizerType; + payload?: Uint8Array; + contentType?: string; + accept?: string; + applicationHeaders?: [string, string][]; + bearerToken?: string; + runtimeSessionId?: string; + mcpSessionId?: string; + mcpProtocolVersion?: string; +}; + +export type GatewayInvokeResponse = { + statusCode: number; + contentType: string; + runtimeSessionId?: string; + mcpSessionId?: string; + mcpProtocolVersion?: string; + requestId?: string; + body: AsyncIterable; +}; + export interface CoreGatewayClient { createGateway(input: CreateGatewayInput, options: CoreOptions): Promise; - getGateway(id: string, options: CoreOptions): Promise; + invokeGateway( + request: GatewayInvokeRequest, + options: CoreOptions, + signal?: AbortSignal, + ): Promise; + getGateway(id: string, options: CoreOptions, signal?: AbortSignal): Promise; listGateways( nextToken: string | undefined, maxResults: number | undefined, diff --git a/src/handlers/runtime/invoke/response.ts b/src/handlers/runtime/invoke/response.ts index d28af16f6..190138c4f 100644 --- a/src/handlers/runtime/invoke/response.ts +++ b/src/handlers/runtime/invoke/response.ts @@ -1,56 +1,16 @@ -import { createWriteStream } from "node:fs"; -import { pipeline } from "node:stream/promises"; import { RuntimeInvokeInterruptedError, RuntimeInvokeResponseError } from "../../../errors"; +import { + classifyStreamingResponse, + writeStreamingResponse, + writeStreamingResponseFile, + type StreamingResponseOutput, +} from "../../../io"; import type { RuntimeInvokeResponse } from "../types"; -interface RuntimeInvokeOutput { - stdout: NodeJS.WriteStream; - stderr: NodeJS.WriteStream; - outputFile?: string; - json?: boolean; - signal?: AbortSignal; -} - const RESPONSE_STREAM_FAILED = "response stream failed"; -const TEXTUAL_SEQUENCE_MEDIA_TYPES = new Set([ - "application/x-ndjson", - "application/ndjson", - "application/json-seq", -]); - -function mediaType(contentType: string): string { - return contentType.split(";", 1)[0]!.trim().toLowerCase(); -} export function classifyRuntimeResponse(contentType: string) { - const type = mediaType(contentType); - if (type === "application/json" || /^application\/[^/]+\+json$/.test(type)) { - return "json"; - } - return type.startsWith("text/") || TEXTUAL_SEQUENCE_MEDIA_TYPES.has(type) ? "text" : "binary"; -} - -async function* countBytes( - body: AsyncIterable, - add: (size: number) => void, -): AsyncGenerator { - for await (const chunk of body) { - const snapshot = Uint8Array.from(chunk); - add(snapshot.byteLength); - yield snapshot; - } -} - -async function writeChunk( - stream: NodeJS.WriteStream, - chunk: string | Uint8Array, - signal?: AbortSignal, -): Promise { - try { - await pipeline([chunk], stream, { end: false, signal }); - } catch (error) { - failure(error); - } + return classifyStreamingResponse(contentType); } export async function writeRuntimeInvokeFile( @@ -59,13 +19,7 @@ export async function writeRuntimeInvokeFile( signal?: AbortSignal, onBytes?: (size: number) => void, ): Promise { - await pipeline( - countBytes(response.body, (size) => { - onBytes?.(size); - }), - createWriteStream(path), - { signal }, - ); + await writeStreamingResponseFile(response, path, signal, onBytes); } function failure(error: unknown): never { @@ -74,46 +28,6 @@ function failure(error: unknown): never { throw new RuntimeInvokeResponseError(RESPONSE_STREAM_FAILED, error); } -async function readBody( - body: AsyncIterable, - signal: AbortSignal | undefined, - onBytes: (size: number) => void, -): Promise { - const chunks: Uint8Array[] = []; - for await (const chunk of body) { - signal?.throwIfAborted(); - const snapshot = Uint8Array.from(chunk); - onBytes(snapshot.byteLength); - chunks.push(snapshot); - } - return Buffer.concat(chunks); -} - -async function writeJsonResponse( - response: RuntimeInvokeResponse, - bytes: Uint8Array, - output: RuntimeInvokeOutput, -): Promise { - const { body: _body, ...responseMetadata } = response; - let bodyEncoding: "utf8" | "base64" = "base64"; - let body = Buffer.from(bytes).toString("base64"); - - if (classifyRuntimeResponse(response.contentType) !== "binary") { - try { - body = new TextDecoder("utf-8", { fatal: true }).decode(bytes); - bodyEncoding = "utf8"; - } catch {} - } - - const envelope = JSON.stringify({ - ...responseMetadata, - bodyEncoding, - body, - complete: true, - }); - await writeChunk(output.stdout, envelope, output.signal); -} - function summary( response: RuntimeInvokeResponse, byteCount: number, @@ -134,50 +48,12 @@ function summary( export async function writeRuntimeInvokeResponse( response: RuntimeInvokeResponse, - output: RuntimeInvokeOutput, + output: StreamingResponseOutput, ): Promise { - if ( - output.outputFile === undefined && - !output.json && - output.stdout.isTTY && - classifyRuntimeResponse(response.contentType) === "binary" - ) { - await writeChunk(output.stderr, summary(response, 0, false)); - throw new TypeError("Binary or unknown response content requires --output-file or --json"); - } - - let byteCount = 0; - try { - if (output.outputFile !== undefined) { - await writeRuntimeInvokeFile( - response, - output.outputFile, - output.signal, - (size) => (byteCount += size), - ); - } else if (output.json) { - const bytes = await readBody(response.body, output.signal, (size) => (byteCount += size)); - await writeJsonResponse(response, bytes, output); - } else { - await pipeline( - countBytes(response.body, (size) => (byteCount += size)), - output.stdout, - { end: false, signal: output.signal }, - ); - } - } catch (error) { - await writeChunk( - output.stderr, - summary( - response, - byteCount, - false, - (error as Error)?.name === "AbortError" ? "interrupted" : "response-stream-failed", - ), - ); - failure(error); - } - if (!output.json) { - await writeChunk(output.stderr, summary(response, byteCount, true)); - } + await writeStreamingResponse(response, output, { + metadata: ({ body: _body, ...metadata }) => metadata, + summary, + fail: failure, + binaryTtyError: "Binary or unknown response content requires --output-file or --json", + }); } diff --git a/src/io/index.ts b/src/io/index.ts index 18ee010ab..7e3d4b635 100644 --- a/src/io/index.ts +++ b/src/io/index.ts @@ -19,5 +19,13 @@ export { } from "./exec"; export { FsReadWriteJson } from "./json"; export { SourceResolver, type SourceResolverConfig } from "./source"; +export { + classifyStreamingResponse, + writeStreamingResponse, + writeStreamingResponseFile, + type StreamingResponse, + type StreamingResponseOutput, + type StreamingResponseWriter, +} from "./streamingResponse"; export type { AppIO, ReadWriteJson } from "./types"; export { warn } from "./warn"; diff --git a/src/io/source.ts b/src/io/source.ts index 2a8ecce08..60ad1e952 100644 --- a/src/io/source.ts +++ b/src/io/source.ts @@ -1,3 +1,4 @@ +import { createReadStream } from "node:fs"; import { readFile } from "node:fs/promises"; import { addAbortSignal } from "node:stream"; import { buffer } from "node:stream/consumers"; @@ -53,9 +54,15 @@ export class SourceResolver { } this.stdinClaimedBy = name; - const stdin = this.config.signal - ? addAbortSignal(this.config.signal, this.config.stdin) - : this.config.stdin; + // Importing Ink under Bun drains the process.stdin stream object before a + // headless handler can claim it, while file descriptor 0 still retains the + // bytes. Use a fresh non-closing stream for the real process input; injected + // AppIO streams continue through the ordinary testable path. + const source = + this.config.stdin === process.stdin + ? createReadStream("", { fd: 0, autoClose: false }) + : this.config.stdin; + const stdin = this.config.signal ? addAbortSignal(this.config.signal, source) : source; try { return await buffer(stdin); } catch (error) { diff --git a/src/io/streamingResponse.ts b/src/io/streamingResponse.ts new file mode 100644 index 000000000..30f8de42f --- /dev/null +++ b/src/io/streamingResponse.ts @@ -0,0 +1,179 @@ +import { createWriteStream } from "node:fs"; +import { pipeline } from "node:stream/promises"; + +export type StreamingResponse = { + contentType: string; + body: AsyncIterable; +}; + +export type StreamingResponseOutput = { + stdout: NodeJS.WriteStream; + stderr: NodeJS.WriteStream; + outputFile?: string; + json?: boolean; + signal?: AbortSignal; +}; + +export type StreamingResponseWriter = { + metadata: (response: T) => Record; + summary: (response: T, byteCount: number, complete: boolean, error?: string) => string; + fail: (error: unknown) => never; + binaryTtyError: string; +}; + +const TEXTUAL_SEQUENCE_MEDIA_TYPES = new Set([ + "application/x-ndjson", + "application/ndjson", + "application/json-seq", +]); + +function mediaType(contentType: string): string { + return contentType.split(";", 1)[0]!.trim().toLowerCase(); +} + +export function classifyStreamingResponse(contentType: string): "json" | "text" | "binary" { + const type = mediaType(contentType); + if (type === "application/json" || /^application\/[^/]+\+json$/.test(type)) { + return "json"; + } + return type.startsWith("text/") || TEXTUAL_SEQUENCE_MEDIA_TYPES.has(type) ? "text" : "binary"; +} + +async function* countBytes( + body: AsyncIterable, + add: (size: number) => void, +): AsyncGenerator { + for await (const chunk of body) { + const snapshot = Uint8Array.from(chunk); + add(snapshot.byteLength); + yield snapshot; + } +} + +async function writeChunk( + stream: NodeJS.WriteStream, + chunk: string | Uint8Array, + signal: AbortSignal | undefined, + fail: (error: unknown) => never, +): Promise { + try { + await pipeline([chunk], stream, { end: false, signal }); + } catch (error) { + fail(error); + } +} + +export async function writeStreamingResponseFile( + response: StreamingResponse, + path: string, + signal?: AbortSignal, + onBytes?: (size: number) => void, +): Promise { + await pipeline( + countBytes(response.body, (size) => { + onBytes?.(size); + }), + createWriteStream(path), + { signal }, + ); +} + +async function readBody( + body: AsyncIterable, + signal: AbortSignal | undefined, + onBytes: (size: number) => void, +): Promise { + const chunks: Uint8Array[] = []; + for await (const chunk of body) { + signal?.throwIfAborted(); + const snapshot = Uint8Array.from(chunk); + onBytes(snapshot.byteLength); + chunks.push(snapshot); + } + return Buffer.concat(chunks); +} + +async function writeJsonResponse( + response: T, + bytes: Uint8Array, + output: StreamingResponseOutput, + writer: StreamingResponseWriter, +): Promise { + let bodyEncoding: "utf8" | "base64" = "base64"; + let body = Buffer.from(bytes).toString("base64"); + + if (classifyStreamingResponse(response.contentType) !== "binary") { + try { + body = new TextDecoder("utf-8", { fatal: true }).decode(bytes); + bodyEncoding = "utf8"; + } catch {} + } + + const envelope = JSON.stringify({ + ...writer.metadata(response), + bodyEncoding, + body, + complete: true, + }); + await writeChunk(output.stdout, envelope, output.signal, writer.fail); +} + +export async function writeStreamingResponse( + response: T, + output: StreamingResponseOutput, + writer: StreamingResponseWriter, +): Promise { + if ( + output.outputFile === undefined && + !output.json && + output.stdout.isTTY && + classifyStreamingResponse(response.contentType) === "binary" + ) { + await writeChunk(output.stderr, writer.summary(response, 0, false), output.signal, writer.fail); + throw new TypeError(writer.binaryTtyError); + } + + let byteCount = 0; + try { + if (output.outputFile !== undefined) { + await writeStreamingResponseFile(response, output.outputFile, output.signal, (size) => { + byteCount += size; + }); + } else if (output.json) { + const bytes = await readBody(response.body, output.signal, (size) => { + byteCount += size; + }); + await writeJsonResponse(response, bytes, output, writer); + } else { + await pipeline( + countBytes(response.body, (size) => { + byteCount += size; + }), + output.stdout, + { end: false, signal: output.signal }, + ); + } + } catch (error) { + await writeChunk( + output.stderr, + writer.summary( + response, + byteCount, + false, + (error as Error)?.name === "AbortError" ? "interrupted" : "response-stream-failed", + ), + output.signal, + writer.fail, + ); + writer.fail(error); + } + + if (!output.json) { + await writeChunk( + output.stderr, + writer.summary(response, byteCount, true), + output.signal, + writer.fail, + ); + } +} diff --git a/src/testing/TestCoreClient.tsx b/src/testing/TestCoreClient.tsx index df3414166..197fba2bf 100644 --- a/src/testing/TestCoreClient.tsx +++ b/src/testing/TestCoreClient.tsx @@ -89,6 +89,8 @@ import type { CreateGatewayInput, CreateGatewayRuleInput, CreateGatewayTargetInput, + GatewayInvokeRequest, + GatewayInvokeResponse, } from "../handlers/gateway/types"; import type { CoreIdentityClient, @@ -232,6 +234,11 @@ const DEFAULT_RUNTIME_INVOKE_RESPONSE: RuntimeInvokeResponse = { contentType: "application/json", body: events([]), }; +const DEFAULT_GATEWAY_INVOKE_RESPONSE: GatewayInvokeResponse = { + statusCode: 200, + contentType: "application/json", + body: events([]), +}; // TestHarnessClient is the harness sub-client of TestCoreClient. export class TestHarnessClient implements CoreHarnessClient { @@ -840,6 +847,7 @@ export class TestGatewayClient implements CoreGatewayClient { private listConnectorResponses = new Map(); private getRuleResponse: GetGatewayRuleResponse = DEFAULT_GET_GATEWAY_RULE_RESPONSE; private listRuleResponses = new Map(); + private invokeResponse: GatewayInvokeResponse = DEFAULT_GATEWAY_INVOKE_RESPONSE; private error?: Error; setGetResponse(response: GetGatewayResponse): this { @@ -882,6 +890,11 @@ export class TestGatewayClient implements CoreGatewayClient { return this; } + setInvokeResponse(response: GatewayInvokeResponse): this { + this.invokeResponse = response; + return this; + } + setError(error: Error | undefined): this { this.error = error; return this; @@ -896,8 +909,25 @@ export class TestGatewayClient implements CoreGatewayClient { return DEFAULT_CREATE_GATEWAY_RESPONSE; } - async getGateway(id: string, options: CoreOptions): Promise { - this.calls.push({ method: "getGateway", args: [id, options] }); + async invokeGateway( + request: GatewayInvokeRequest, + options: CoreOptions, + signal?: AbortSignal, + ): Promise { + this.calls.push({ method: "invokeGateway", args: [request, options, signal] }); + if (this.error) throw this.error; + return this.invokeResponse; + } + + async getGateway( + id: string, + options: CoreOptions, + signal?: AbortSignal, + ): Promise { + this.calls.push({ + method: "getGateway", + args: [id, options, ...(signal ? [signal] : [])], + }); if (this.error) throw this.error; return this.getResponse; }