From e04360f83e9a2937f94d7fc47e1f83c1715f9f4a Mon Sep 17 00:00:00 2001 From: andypalmi Date: Mon, 3 Aug 2026 16:12:33 +0200 Subject: [PATCH 1/2] feat(mcp): shared query schema fragments for platform tools Add forge/ee/lib/mcp/schemas.js, a shared module of composable zod fragments the platform read tools import instead of redefining entity-id and pagination/search/sort/audit-log query fields in each tool file. - entity-id params: teamId, applicationId, hostedInstanceId (UUID), remoteInstanceId, snapshotId - query fragments composed per route by spreading only the params the backing finder honors: cursorParam/limitParam (basePagination), pageParam, searchQuery, sortParams, auditLogFilters - appendQuery serialises a tool's supported params onto the request URL The module lives one level above tools/ so the tool loader does not register it as a tool module. Closes #7669 --- forge/ee/lib/mcp/schemas.js | 92 ++++++++++++++++++++++ test/unit/forge/ee/lib/mcp/schemas_spec.js | 76 ++++++++++++++++++ 2 files changed, 168 insertions(+) create mode 100644 forge/ee/lib/mcp/schemas.js create mode 100644 test/unit/forge/ee/lib/mcp/schemas_spec.js diff --git a/forge/ee/lib/mcp/schemas.js b/forge/ee/lib/mcp/schemas.js new file mode 100644 index 0000000000..03d628a008 --- /dev/null +++ b/forge/ee/lib/mcp/schemas.js @@ -0,0 +1,92 @@ +const { z } = require('zod') + +// Hosted instances are Projects (UUID primary key); the other entities use hashids. +const teamId = z.string().describe('The ID or hashid of the team') +const applicationId = z.string().describe('The ID or hashid of the application') +const hostedInstanceId = z.string().uuid().describe('The UUID of the hosted instance') +const remoteInstanceId = z.string().describe('The ID or hashid of the remote instance') +const snapshotId = z.string().describe('The hashid of the snapshot') + +// Query fragments composed per tool by spreading only the ones the backing +// route's finder actually honors. Not the same as the route's declared query +// schema: most list routes reuse a generic PaginationParams that advertises +// page/sort/dir/order even when the finder ignores them. Match the finder. + +const cursorParam = { + cursor: z.string().optional().describe('Opaque cursor from a previous page') +} +const limitParam = { + limit: z.number().int().min(1).max(50).default(10).describe('Maximum number of records to return (1-50, default 10)') +} +const basePagination = { ...cursorParam, ...limitParam } + +// Only Device.getAll and Project.byTeam read page and compute an offset. +const pageParam = { + page: z.number().int().min(1).optional().default(1).describe('1-based page number (offset pagination)') +} + +const searchQuery = { + query: z.string().optional().describe('Free-text search filter') +} + +// Only Project.byTeam honors sort; no finder reads the legacy `order` alias. +const sortParams = { + sort: z.string().optional().describe('Field name to sort by'), + dir: z.enum(['asc', 'desc']).optional().describe('Sort direction') +} + +// scope is route-specific (its enum differs per entity), so each tool declares it inline. +const auditLogFilters = { + event: z.union([z.string(), z.array(z.string())]).optional().describe('Filter by audit event name, or an array of event names'), + username: z.string().optional().describe('Filter by the username that triggered the event') +} + +const cursorParamKeys = Object.keys(cursorParam) +const limitParamKeys = Object.keys(limitParam) +const basePaginationKeys = Object.keys(basePagination) +const pageParamKeys = Object.keys(pageParam) +const searchQueryKeys = Object.keys(searchQuery) +const sortParamsKeys = Object.keys(sortParams) +const auditLogFilterKeys = Object.keys(auditLogFilters) + +// Serialise the given query keys from args onto a url: only defined values, +// URL-encoded, an array value appended once per element. +function appendQuery (url, args, keys) { + const params = new URLSearchParams() + for (const key of keys) { + const value = args[key] + if (value === undefined || value === null) { + continue + } + if (Array.isArray(value)) { + value.forEach(v => params.append(key, v)) + } else { + params.append(key, value) + } + } + const queryString = params.toString() + return queryString ? `${url}?${queryString}` : url +} + +module.exports = { + teamId, + applicationId, + hostedInstanceId, + remoteInstanceId, + snapshotId, + cursorParam, + limitParam, + basePagination, + pageParam, + searchQuery, + sortParams, + auditLogFilters, + cursorParamKeys, + limitParamKeys, + basePaginationKeys, + pageParamKeys, + searchQueryKeys, + sortParamsKeys, + auditLogFilterKeys, + appendQuery +} diff --git a/test/unit/forge/ee/lib/mcp/schemas_spec.js b/test/unit/forge/ee/lib/mcp/schemas_spec.js new file mode 100644 index 0000000000..ae9d953474 --- /dev/null +++ b/test/unit/forge/ee/lib/mcp/schemas_spec.js @@ -0,0 +1,76 @@ +const should = require('should') // eslint-disable-line no-unused-vars + +const FF_UTIL = require('flowforge-test-utils') + +const { + teamId, + applicationId, + hostedInstanceId, + remoteInstanceId, + snapshotId, + basePagination, + cursorParamKeys, + limitParamKeys, + basePaginationKeys, + pageParamKeys, + searchQueryKeys, + sortParamsKeys, + auditLogFilterKeys, + appendQuery +} = FF_UTIL.require('forge/ee/lib/mcp/schemas') + +describe('MCP shared query schemas', function () { + describe('entity-id params', function () { + it('hashid params accept any non-empty string', function () { + for (const param of [teamId, applicationId, remoteInstanceId, snapshotId]) { + param.parse('aBc123').should.equal('aBc123') + } + }) + it('hostedInstanceId accepts a UUID and rejects a hashid', function () { + hostedInstanceId.parse('4f8c1e2a-1b2c-4d3e-8f90-abcdef123456').should.be.a.String() + hostedInstanceId.safeParse('aBc123').success.should.equal(false) + }) + }) + + describe('fragment composition', function () { + it('basePagination is cursor plus limit only (no page)', function () { + basePaginationKeys.should.eql(['cursor', 'limit']) + }) + it('page, search and sort are separate opt-in fragments', function () { + cursorParamKeys.should.eql(['cursor']) + limitParamKeys.should.eql(['limit']) + pageParamKeys.should.eql(['page']) + searchQueryKeys.should.eql(['query']) + sortParamsKeys.should.eql(['sort', 'dir']) + auditLogFilterKeys.should.eql(['event', 'username']) + }) + it('key list matches the shape object it describes', function () { + basePaginationKeys.should.eql(Object.keys(basePagination)) + }) + }) + + describe('appendQuery', function () { + it('returns the url unchanged when no supported params are set', function () { + appendQuery('/api/v1/x', {}, basePaginationKeys).should.equal('/api/v1/x') + }) + it('returns the url unchanged when only unsupported keys are present', function () { + appendQuery('/api/v1/x', { teamId: 'abc' }, basePaginationKeys).should.equal('/api/v1/x') + }) + it('serialises only the defined, supported params', function () { + appendQuery('/api/v1/x', { limit: 10, page: 2, cursor: undefined }, [...basePaginationKeys, ...pageParamKeys]) + .should.equal('/api/v1/x?limit=10&page=2') + }) + it('url-encodes values', function () { + appendQuery('/api/v1/x', { query: 'a b&c' }, searchQueryKeys) + .should.equal('/api/v1/x?query=a+b%26c') + }) + it('appends an array value once per element', function () { + appendQuery('/api/v1/x', { event: ['flows.created', 'flows.deleted'] }, auditLogFilterKeys) + .should.equal('/api/v1/x?event=flows.created&event=flows.deleted') + }) + it('skips null and undefined but keeps other values', function () { + appendQuery('/api/v1/x', { limit: null, page: 1 }, [...basePaginationKeys, ...pageParamKeys]) + .should.equal('/api/v1/x?page=1') + }) + }) +}) From ae8469b5abc65853ec6ccfcda3a3d064de0f87df Mon Sep 17 00:00:00 2001 From: andypalmi Date: Mon, 3 Aug 2026 17:27:16 +0200 Subject: [PATCH 2/2] feat(mcp): add broker read tools Expose read-only team broker tools: list/get clients, list/get brokers, list topics and get the AsyncAPI schema. --- forge/ee/lib/mcp/tools/broker.js | 113 ++++++++++++ forge/routes/auth/permissions.js | 6 +- .../forge/ee/lib/mcp/tools/broker_spec.js | 164 ++++++++++++++++++ 3 files changed, 282 insertions(+), 1 deletion(-) create mode 100644 forge/ee/lib/mcp/tools/broker.js create mode 100644 test/unit/forge/ee/lib/mcp/tools/broker_spec.js diff --git a/forge/ee/lib/mcp/tools/broker.js b/forge/ee/lib/mcp/tools/broker.js new file mode 100644 index 0000000000..21c86f8172 --- /dev/null +++ b/forge/ee/lib/mcp/tools/broker.js @@ -0,0 +1,113 @@ +const { z } = require('zod') + +const { teamId, basePagination, basePaginationKeys, searchQuery, appendQuery } = require('../schemas') + +module.exports = [ + { + name: 'platform_list_broker_clients', + title: 'List Broker Clients', + description: `FlowFuse platform automation tool: + Lists the MQTT clients registered on the team broker (the built-in MQTT broker that ships with the platform). + Each entry identifies the client username and, where known, the hosted instance or remote instance it belongs to. This does not include MQTT credentials. + Supports username search and pagination. + This tool requires the enterprise license tier and the team broker feature enabled for the team; if the team does not have it enabled, the request returns a not found response.`, + annotations: { readOnlyHint: true, destructiveHint: false }, + inputSchema: { + teamId, + ...basePagination, + ...searchQuery + }, + handler: async (args, { inject }) => { + const url = appendQuery(`/api/v1/teams/${args.teamId}/broker/clients`, args, [...basePaginationKeys, 'query']) + const response = await inject({ method: 'GET', url }) + return response + } + }, + { + name: 'platform_get_broker_client', + title: 'Get Broker Client', + description: `FlowFuse platform automation tool: + Gets a single MQTT client registered on the team broker, identified by its username. This does not include MQTT credentials. + Use this after platform_list_broker_clients to inspect one client in detail. + This tool requires the enterprise license tier and the team broker feature enabled for the team; if the team does not have it enabled, the request returns a not found response.`, + annotations: { readOnlyHint: true, destructiveHint: false }, + inputSchema: { + teamId, + username: z.string().describe('Username of the broker client to fetch') + }, + handler: async (args, { inject }) => { + const response = await inject({ method: 'GET', url: `/api/v1/teams/${args.teamId}/broker/client/${args.username}` }) + return response + } + }, + { + name: 'platform_list_brokers', + title: 'List Brokers', + description: `FlowFuse platform automation tool: + Lists the brokers configured for a team: the built-in team broker plus any 3rd-party MQTT brokers that have been linked to the team. This does not include MQTT credentials. + Use this to find a broker's ID before calling platform_get_broker, platform_list_broker_topics, or platform_get_broker_schema. + Supports pagination. + This tool requires the enterprise license tier and the team broker feature enabled for the team; if the team does not have it enabled, the request returns a not found response.`, + annotations: { readOnlyHint: true, destructiveHint: false }, + inputSchema: { + teamId, + ...basePagination + }, + handler: async (args, { inject }) => { + const url = appendQuery(`/api/v1/teams/${args.teamId}/brokers`, args, basePaginationKeys) + const response = await inject({ method: 'GET', url }) + return response + } + }, + { + name: 'platform_get_broker', + title: 'Get Broker', + description: `FlowFuse platform automation tool: + Gets the details and status of a single broker: the built-in team broker or a linked 3rd-party MQTT broker. This does not include MQTT credentials. + Use this after platform_list_brokers to inspect one broker in detail. + This tool requires the enterprise license tier and the team broker feature enabled for the team; if the team does not have it enabled, the request returns a not found response.`, + annotations: { readOnlyHint: true, destructiveHint: false }, + inputSchema: { + teamId, + brokerId: z.string().describe("broker id: either the literal 'team-broker' or a 3rd-party broker hashid") + }, + handler: async (args, { inject }) => { + const response = await inject({ method: 'GET', url: `/api/v1/teams/${args.teamId}/brokers/${args.brokerId}` }) + return response + } + }, + { + name: 'platform_list_broker_topics', + title: 'List Broker Topics', + description: `FlowFuse platform automation tool: + Lists the MQTT topics that have been observed on a broker, along with any recorded metadata and inferred payload schema for each topic. + Use this to understand what data is flowing through a broker before wiring up new flows that publish or subscribe to it. + This tool requires the enterprise license tier and the team broker feature enabled for the team; if the team does not have it enabled, the request returns a not found response.`, + annotations: { readOnlyHint: true, destructiveHint: false }, + inputSchema: { + teamId, + brokerId: z.string().describe("broker id: either the literal 'team-broker' or a 3rd-party broker hashid") + }, + handler: async (args, { inject }) => { + const response = await inject({ method: 'GET', url: `/api/v1/teams/${args.teamId}/brokers/${args.brokerId}/topics` }) + return response + } + }, + { + name: 'platform_get_broker_schema', + title: 'Get Broker Schema', + description: `FlowFuse platform automation tool: + Gets the auto-generated AsyncAPI topic schema for a broker, built from the topics observed on it. + Use this when the user wants a documented overview of a broker's topic structure and message shapes, for example to share with another team or to generate integration code. + This tool requires the enterprise license tier and the team broker feature enabled for the team; if the team does not have it enabled, the request returns a not found response.`, + annotations: { readOnlyHint: true, destructiveHint: false }, + inputSchema: { + teamId, + brokerId: z.string().describe("broker id: either the literal 'team-broker' or a 3rd-party broker hashid") + }, + handler: async (args, { inject }) => { + const response = await inject({ method: 'GET', url: `/api/v1/teams/${args.teamId}/broker/${args.brokerId}/schema` }) + return response + } + } +] diff --git a/forge/routes/auth/permissions.js b/forge/routes/auth/permissions.js index c1740abaa9..ee75a6c5a5 100644 --- a/forge/routes/auth/permissions.js +++ b/forge/routes/auth/permissions.js @@ -82,7 +82,11 @@ const IMPLICIT_TOKEN_SCOPES = { 'stack:list', 'flow-blueprint:list', 'project:status', - 'template:list' + 'template:list', + // broker + 'broker:clients:list', // list/get team broker clients + 'broker:credentials:list', // list/get brokers + 'broker:topics:list' // list broker topics, get broker schema ] } diff --git a/test/unit/forge/ee/lib/mcp/tools/broker_spec.js b/test/unit/forge/ee/lib/mcp/tools/broker_spec.js new file mode 100644 index 0000000000..cb3c860697 --- /dev/null +++ b/test/unit/forge/ee/lib/mcp/tools/broker_spec.js @@ -0,0 +1,164 @@ +const should = require('should') // eslint-disable-line no-unused-vars +const sinon = require('sinon') + +const tools = require('../../../../../../../forge/ee/lib/mcp/tools/broker') + +function getTool (name) { + return tools.find(tool => tool.name === name) +} + +describe('MCP Broker Tools', function () { + let inject + + beforeEach(function () { + inject = sinon.stub() + }) + + describe('platform_list_broker_clients', function () { + const tool = getTool('platform_list_broker_clients') + + it('injects the broker clients list route and returns the response', async function () { + const routeResponse = { statusCode: 200, json: () => ({ clients: [], count: 0 }) } + inject.withArgs({ method: 'GET', url: '/api/v1/teams/team1/broker/clients' }).resolves(routeResponse) + + const response = await tool.handler({ teamId: 'team1' }, { inject }) + + inject.calledOnce.should.be.true() + response.should.equal(routeResponse) + }) + + it('serialises pagination and search params', async function () { + inject.resolves({ statusCode: 200, json: () => ({ clients: [] }) }) + + await tool.handler({ teamId: 'team1', cursor: 'abc', limit: 20, query: 'sensor' }, { inject }) + + inject.firstCall.args[0].url.should.equal('/api/v1/teams/team1/broker/clients?cursor=abc&limit=20&query=sensor') + }) + + it('passes through an error response', async function () { + const errorResponse = { statusCode: 404, json: () => ({ code: 'not_found' }) } + inject.resolves(errorResponse) + + const response = await tool.handler({ teamId: 'team1' }, { inject }) + response.should.equal(errorResponse) + }) + }) + + describe('platform_get_broker_client', function () { + const tool = getTool('platform_get_broker_client') + + it('injects the broker client route for the given username', async function () { + const routeResponse = { statusCode: 200, json: () => ({ username: 'client1' }) } + inject.withArgs({ method: 'GET', url: '/api/v1/teams/team1/broker/client/client1' }).resolves(routeResponse) + + const response = await tool.handler({ teamId: 'team1', username: 'client1' }, { inject }) + + inject.calledOnce.should.be.true() + response.should.equal(routeResponse) + }) + + it('passes through an error response', async function () { + const errorResponse = { statusCode: 404, json: () => ({ code: 'not_found' }) } + inject.resolves(errorResponse) + + const response = await tool.handler({ teamId: 'team1', username: 'client1' }, { inject }) + response.should.equal(errorResponse) + }) + }) + + describe('platform_list_brokers', function () { + const tool = getTool('platform_list_brokers') + + it('injects the brokers list route and returns the response', async function () { + const routeResponse = { statusCode: 200, json: () => ({ brokers: [] }) } + inject.withArgs({ method: 'GET', url: '/api/v1/teams/team1/brokers' }).resolves(routeResponse) + + const response = await tool.handler({ teamId: 'team1' }, { inject }) + + inject.calledOnce.should.be.true() + response.should.equal(routeResponse) + }) + + it('serialises pagination params', async function () { + inject.resolves({ statusCode: 200, json: () => ({ brokers: [] }) }) + + await tool.handler({ teamId: 'team1', cursor: 'abc', limit: 5 }, { inject }) + + inject.firstCall.args[0].url.should.equal('/api/v1/teams/team1/brokers?cursor=abc&limit=5') + }) + + it('passes through an error response', async function () { + const errorResponse = { statusCode: 404, json: () => ({ code: 'not_found' }) } + inject.resolves(errorResponse) + + const response = await tool.handler({ teamId: 'team1' }, { inject }) + response.should.equal(errorResponse) + }) + }) + + describe('platform_get_broker', function () { + const tool = getTool('platform_get_broker') + + it('injects the broker detail route for the given broker', async function () { + const routeResponse = { statusCode: 200, json: () => ({ id: 'team-broker' }) } + inject.withArgs({ method: 'GET', url: '/api/v1/teams/team1/brokers/team-broker' }).resolves(routeResponse) + + const response = await tool.handler({ teamId: 'team1', brokerId: 'team-broker' }, { inject }) + + inject.calledOnce.should.be.true() + response.should.equal(routeResponse) + }) + + it('passes through an error response', async function () { + const errorResponse = { statusCode: 404, json: () => ({ code: 'not_found' }) } + inject.resolves(errorResponse) + + const response = await tool.handler({ teamId: 'team1', brokerId: 'team-broker' }, { inject }) + response.should.equal(errorResponse) + }) + }) + + describe('platform_list_broker_topics', function () { + const tool = getTool('platform_list_broker_topics') + + it('injects the broker topics route for the given broker', async function () { + const routeResponse = { statusCode: 200, json: () => ({ topics: [] }) } + inject.withArgs({ method: 'GET', url: '/api/v1/teams/team1/brokers/team-broker/topics' }).resolves(routeResponse) + + const response = await tool.handler({ teamId: 'team1', brokerId: 'team-broker' }, { inject }) + + inject.calledOnce.should.be.true() + response.should.equal(routeResponse) + }) + + it('passes through an error response', async function () { + const errorResponse = { statusCode: 404, json: () => ({ code: 'not_found' }) } + inject.resolves(errorResponse) + + const response = await tool.handler({ teamId: 'team1', brokerId: 'team-broker' }, { inject }) + response.should.equal(errorResponse) + }) + }) + + describe('platform_get_broker_schema', function () { + const tool = getTool('platform_get_broker_schema') + + it('injects the broker schema route for the given broker', async function () { + const routeResponse = { statusCode: 200, json: () => ({ channels: {} }) } + inject.withArgs({ method: 'GET', url: '/api/v1/teams/team1/broker/team-broker/schema' }).resolves(routeResponse) + + const response = await tool.handler({ teamId: 'team1', brokerId: 'team-broker' }, { inject }) + + inject.calledOnce.should.be.true() + response.should.equal(routeResponse) + }) + + it('passes through an error response', async function () { + const errorResponse = { statusCode: 404, json: () => ({ code: 'not_found' }) } + inject.resolves(errorResponse) + + const response = await tool.handler({ teamId: 'team1', brokerId: 'team-broker' }, { inject }) + response.should.equal(errorResponse) + }) + }) +})