feat(mcp-server): dispatch mounted tool calls to the agent in-process

packages/agent-client/src/http-requester.ts

@@ -16,7 +16,7 @@ function parseJson(text: string | undefined): unknown {
 }
 
 export default class HttpRequester {
-  private readonly deserializer: Deserializer;
+  protected readonly deserializer: Deserializer;
 
   private get baseUrl() {
     const prefix = this.options.prefix ? `/${this.options.prefix}` : '';
@@ -31,6 +31,28 @@ export default class HttpRequester {
     this.deserializer = new Deserializer({ keyForAttribute: 'camelCase' });
   }
 
+  // Any transport reaching the agent must build this exact AgentHttpError shape — the approval-gate
+  // detection in domains/action.ts routes on it. Shared with in-process transports via subclassing.
+  protected buildError(status: number, body: unknown, text?: string): AgentHttpError {
+    const hasBody = !!body && typeof body === 'object' && Object.keys(body).length > 0;
+
+    return new AgentHttpError(status, hasBody ? body : parseJson(text), text);
+  }
+
+  protected async deserialize<Data>(
+    body: unknown,
+    text: string | undefined,
+    skipDeserialization?: boolean,
+  ): Promise<Data> {
+    if (skipDeserialization) return body as Data;
+
+    try {
+      return (await this.deserializer.deserialize(body)) as Data;
+    } catch {
+      return (body ?? text) as Data;
+    }
+  }
+
   async query<Data = unknown>({
     method,
     path,
@@ -62,25 +84,15 @@ export default class HttpRequester {
 
       const response = await req;
 
-      if (skipDeserialization) {
-        return response.body as Data;
-      }
-
-      try {
-        return (await this.deserializer.deserialize(response.body)) as Data;
-      } catch {
-        return (response.body ?? response.text) as Data;
-      }
+      return await this.deserialize<Data>(response.body, response.text, skipDeserialization);
     } catch (error) {
       const res = (error as { response?: { status?: number; body?: unknown; text?: string } })
         .response;
       if (!res) throw error; // network/timeout/abort → no HTTP response, propagate raw
 
       const text = typeof res.text === 'string' ? res.text : undefined;
-      const hasBody =
-        !!res.body && typeof res.body === 'object' && Object.keys(res.body).length > 0;
 
-      throw new AgentHttpError(res.status ?? 0, hasBody ? res.body : parseJson(text), text);
+      throw this.buildError(res.status ?? 0, res.body, text);
     }
   }
 

packages/agent-client/src/index.ts

@@ -46,8 +46,10 @@ export function createRemoteAgentClient(params: {
    * agent (e.g. tests). `serverUrl` is the Forest server, distinct from the agent `url` above.
    */
   forestServer?: { serverUrl: string; serverToken: string; renderingId: number | string };
+  httpRequester?: HttpRequester;
 }) {
-  const httpRequester = new HttpRequester(params.token, { url: params.url });
+  const httpRequester =
+    params.httpRequester ?? new HttpRequester(params.token, { url: params.url });
 
   return new RemoteAgentClient({
     actionEndpoints: params.actionEndpoints,

packages/agent/package.json

@@ -29,6 +29,7 @@
     "jsonwebtoken": "^9.0.3",
     "koa": "^3.0.1",
     "koa-jwt": "^4.0.4",
+    "light-my-request": "^5.14.0",
     "luxon": "^3.2.1",
     "object-hash": "^3.0.0",
     "superagent": "^10.3.0",

packages/agent/src/agent.ts

@@ -244,6 +244,10 @@ export default class Agent<S extends TSchema = TSchema> extends FrameworkMounter
    * Enable MCP (Model Context Protocol) server support.
    * This allows AI assistants to interact with your Forest Admin data.
    *
+   * Tool calls reach your data in-process (no socket), so they skip any host-app middleware you
+   * mounted in front of the agent (rate limiting, request logging, WAF). The agent's own JWT auth
+   * and permission checks still run on them, and the inbound MCP request itself is unaffected.
+   *
    * @see {@link https://docs.forestadmin.com/developer-guide-agents-nodejs/agent-customization/ai/mcp-server}
    * @example
    * agent.mountAiMcpServer();
@@ -425,12 +429,17 @@ export default class Agent<S extends TSchema = TSchema> extends FrameworkMounter
         forestServerClient,
         enabledTools: this.mcpEnabledTools,
         basePath: this.mcpBasePath,
+        agentDispatcher: this.getInProcessDispatcher(),
       });
 
       const httpCallback = await mcpServer.getHttpCallback();
       const isMcpRoute = makeIsMcpRoute(this.mcpBasePath);
 
       mcpLogger('Info', 'Server initialized successfully');
+      mcpLogger(
+        'Info',
+        'Tool calls dispatch in-process and skip any middleware mounted in front of the agent',
+      );
 
       return { httpCallback, isMcpRoute };
     } catch (error) {

packages/agent/src/framework-mounter.ts

   for (const task of this.onEachStart) await task(router); // eslint-disable-line no-await-in-loop 
 protected async remount(router: Router): Promise<void> { 
   for (const task of this.onEachStart) await task(router); // eslint-disable-line no-await-in-loop 
@@ -9,6 +9,7 @@ import Koa from 'koa';
 import path from 'path';
 
 import FastifyAdapter from './fastify-adapter';
+import InProcessDispatcher from './mcp-in-process-dispatcher';
 import McpMiddleware from './mcp-middleware';
 
 export default class FrameworkMounter {
@@ -24,6 +25,8 @@ export default class FrameworkMounter {
 
   private readonly fastifyAdapter: FastifyAdapter;
   private readonly mcpMiddleware: McpMiddleware;
+  private readonly inProcessDispatcher: InProcessDispatcher;
+  private inProcessHookRegistered = false;
 
   /** Compute the prefix that the main router should be mounted at in the client's application */
   private get completeMountPrefix(): string {
@@ -35,6 +38,7 @@ export default class FrameworkMounter {
     this.logger = logger;
     this.fastifyAdapter = new FastifyAdapter(logger);
     this.mcpMiddleware = new McpMiddleware();
+    this.inProcessDispatcher = new InProcessDispatcher(logger);
   }
 
   /**
@@ -44,6 +48,23 @@ export default class FrameworkMounter {
     this.mcpMiddleware.setCallback(callback, routeMatcher);
   }
 
+  /**
+   * Dispatcher that runs agent-client requests against the agent's own `/forest` stack in-memory.
+   * Its handler is rebuilt on every (re)mount; the `/forest` prefix is fixed (not the external
+   * mount prefix) because agent-client hardcodes `/forest/...` paths.
+   */
+  protected getInProcessDispatcher(): InProcessDispatcher {
+    if (!this.inProcessHookRegistered) {
+      this.inProcessHookRegistered = true;
+      this.onEachStart.push(async driverRouter => {
+        const router = new Router({ prefix: '/forest' }).use(driverRouter.routes());
+        this.inProcessDispatcher.setHandler(new Koa().use(router.routes()).callback());
+      });
+    }
+
+    return this.inProcessDispatcher;
+  }
+
   protected async mount(router: Router): Promise<void> {
     for (const task of this.onFirstStart) await task(); // eslint-disable-line no-await-in-loop
 

packages/agent/src/mcp-in-process-dispatcher.ts

@@ -0,0 +1,118 @@
+import type { Logger } from '@forestadmin/datasource-toolkit';
+import type {
+  InProcessAgentDispatcher,
+  InProcessDispatchRequest,
+  InProcessDispatchResponse,
+} from '@forestadmin/mcp-server';
+import type { RequestListener } from 'http';
+
+import inject, { type InjectOptions, type InjectPayload } from 'light-my-request';
+
+// Single-sourced from mcp-server (the dispatcher contract) — no re-declaration to drift.
+export type InProcessRequest = InProcessDispatchRequest;
+export type InProcessResponse = InProcessDispatchResponse;
+
+// Matches superagent's HTTP path, which times out after 10s (HttpRequester.query).
+const DEFAULT_TIMEOUT_MS = 10_000;
+
+function toStringQuery(query: Record<string, unknown>): Record<string, string> {
+  const result: Record<string, string> = {};
+
+  for (const [key, value] of Object.entries(query)) {
+    if (value !== undefined) result[key] = String(value);
+  }
+
+  return result;
+}
+
+/**
+ * Dispatches an agent-client request into the agent's own Koa stack in-memory (no socket), so a
+ * mounted MCP server's tool calls run through the real auth/permission/logging middleware.
+ * `setHandler` is refreshed on every (re)mount, so a captured reference never goes stale.
+ */
+export default class InProcessDispatcher implements InProcessAgentDispatcher {
+  private handler: RequestListener | null = null;
+
+  constructor(private readonly logger?: Logger) {}
+
+  setHandler(handler: RequestListener | null): void {
+    this.handler = handler;
+  }
+
+  async request({
+    method,
+    path,
+    headers,
+    query,
+    payload,
+    timeoutMs = DEFAULT_TIMEOUT_MS,
+  }: InProcessRequest): Promise<InProcessResponse> {
+    if (!this.handler) {
+      throw new Error('Cannot dispatch to the agent in-process: it is not mounted yet.');
+    }
+
+    // No socket to abort, so a hung agent handler would hang the tool call forever without this.
+    const response = await this.injectWithTimeout(
+      {
+        method: method.toUpperCase() as InjectOptions['method'],
+        url: path,
+        headers,
+        ...(query && { query: toStringQuery(query) }),
+        ...(payload !== undefined && { payload: payload as InjectPayload }),
+      },
+      timeoutMs,
+    );
+
+    return {
+      status: response.statusCode,
+      body: this.parseBody(response.payload, response.statusCode),
+      text: response.payload,
+    };
+  }
+
+  private async injectWithTimeout(options: InjectOptions, timeoutMs: number) {
+    let timer: NodeJS.Timeout | undefined;
+    const timeout = new Promise<never>((_, reject) => {
+      timer = setTimeout(
+        () => reject(new Error(`In-process dispatch timed out after ${timeoutMs}ms`)),
+        timeoutMs,
+      );
+    });
+
+    const injection = inject(this.handler as RequestListener, options);
+    // If the timeout wins the race, `injection` is left unobserved; swallow+log a late settlement so
+    // a handler that rejects after the timeout can't surface as an unhandledRejection.
+    injection.catch(error =>
+      this.logger?.(
+        'Warn',
+        `In-process handler settled after the ${timeoutMs}ms timeout: ${error}`,
+      ),
+    );
+
+    try {
+      return await Promise.race([injection, timeout]);
+    } finally {
+      if (timer) clearTimeout(timer);
+    }
+  }
+
+  private parseBody(payload: string, status: number): unknown {
+    if (payload === '') return undefined;
+
+    try {
+      return JSON.parse(payload);
+    } catch {
+      const preview = payload.length > 200 ? `${payload.slice(0, 200)}…` : payload;
+      const message = `In-process dispatch received a non-JSON ${status} response body: ${preview}`;
+
+      // A 2xx must carry a JSON:API body here; a non-JSON success body means misbehaving middleware
+      // (HTML, redirect) — fail loudly rather than hand the tool `undefined`. On 4xx+ the raw text
+      // is preserved by the error path (buildError falls back to parseJson(text)), so drop the body.
+      if (status < 400) throw new Error(message);
+
+      this.logger?.('Warn', message);
+
+      return undefined;
+    }
+  }
+}

packages/agent/test/agent-integration.test.ts

@@ -594,6 +594,38 @@ describe('Agent Integration Tests', () => {
           lastName: 'Wilson',
         });
       });
+
+      it('dispatches the tool call to the agent in-process, not over a socket', async () => {
+        const dispatcher = (
+          testContext.agent as unknown as {
+            getInProcessDispatcher: () => { request: (...args: unknown[]) => Promise<unknown> };
+          }
+        ).getInProcessDispatcher();
+        const requestSpy = jest.spyOn(dispatcher, 'request');
+
+        try {
+          const token = createTestToken({ scopes: ['mcp:read'], renderingId: 1 });
+
+          const response = await superagent
+            .post(`${testContext.baseUrl}/mcp`)
+            .set('Authorization', `Bearer ${token}`)
+            .set('Content-Type', 'application/json')
+            .set('Accept', 'text/event-stream, application/json')
+            .send({
+              jsonrpc: '2.0',
+              method: 'tools/call',
+              id: 3,
+              params: { name: 'list', arguments: { collectionName: 'users' } },
+            });
+
+          expect(response.status).toBe(200);
+          expect(requestSpy).toHaveBeenCalledWith(
+            expect.objectContaining({ method: 'get', path: '/forest/users' }),
+          );
+        } finally {
+          requestSpy.mockRestore();
+        }
+      });
     });
 
     describe('Routes coexistence', () => {

packages/agent/test/agent.test.ts

@@ -591,6 +591,20 @@ describe('Agent', () => {
       expect(mcpServerSpy).toHaveBeenCalledWith(expect.objectContaining({ basePath: '/mcp' }));
     });
 
+    test('passes an in-process agentDispatcher to ForestMCPServer', async () => {
+      const options = factories.forestAdminHttpDriverOptions.build();
+      const agent = new Agent(options);
+
+      agent.mountAiMcpServer();
+      await agent.start();
+
+      expect(mcpServerSpy).toHaveBeenCalledWith(
+        expect.objectContaining({
+          agentDispatcher: expect.objectContaining({ request: expect.any(Function) }),
+        }),
+      );
+    });
+
     test('threads a basePath-scoped route matcher to the MCP middleware', async () => {
       const options = factories.forestAdminHttpDriverOptions.build();
       const agent = new Agent(options);

packages/agent/test/mcp-in-process-dispatcher.rejection.test.ts

@@ -0,0 +1,39 @@
+import type { RequestListener } from 'http';
+
+const mockInject = jest.fn();
+
+jest.mock('light-my-request', () => ({
+  __esModule: true,
+  default: (...args: unknown[]) => mockInject(...args),
+}));
+
+// eslint-disable-next-line import/first
+import InProcessDispatcher from '../src/mcp-in-process-dispatcher';
+
+describe('InProcessDispatcher — injection rejection safety', () => {
+  it('observes a late injection rejection (logs it) instead of leaking an unhandled rejection', async () => {
+    const logger = jest.fn();
+    // inject() rejects after the timeout has already won the race — the losing promise must not
+    // become an unhandledRejection.
+    mockInject.mockReturnValue(
+      new Promise((_resolve, reject) => {
+        setTimeout(() => reject(new Error('inject blew up')), 30);
+      }),
+    );
+    const dispatcher = new InProcessDispatcher(logger);
+    dispatcher.setHandler((() => {}) as unknown as RequestListener);
+
+    await expect(
+      dispatcher.request({ method: 'get', path: '/x', headers: {}, timeoutMs: 10 }),
+    ).rejects.toThrow(/timed out after 10ms/);
+
+    await new Promise(resolve => {
+      setTimeout(resolve, 40);
+    });
+
+    expect(logger).toHaveBeenCalledWith(
+      'Warn',
+      expect.stringContaining('settled after the 10ms timeout'),
+    );
+  });
+});