feat: add external MCP server support with streamable-http proxy
Support non-containerized MCP servers via externalUrl field and add streamable-http session management for HA MCP proof of concept. - Add externalUrl, command, containerPort fields to McpServer schema - Skip Docker orchestration for external servers (virtual instances) - Implement streamable-http proxy with Mcp-Session-Id session management - Parse SSE-framed responses from streamable-http endpoints - Add command passthrough to Docker container creation - Create HA MCP example manifest (examples/ha-mcp.yaml) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
26
examples/ha-mcp.yaml
Normal file
26
examples/ha-mcp.yaml
Normal file
@@ -0,0 +1,26 @@
|
|||||||
|
servers:
|
||||||
|
- name: ha-mcp
|
||||||
|
description: "Home Assistant MCP - smart home control via MCP"
|
||||||
|
dockerImage: "ghcr.io/homeassistant-ai/ha-mcp:2.4"
|
||||||
|
transport: STREAMABLE_HTTP
|
||||||
|
containerPort: 3000
|
||||||
|
# For mcpd-managed containers:
|
||||||
|
command:
|
||||||
|
- python
|
||||||
|
- "-c"
|
||||||
|
- "from ha_mcp.server import HomeAssistantSmartMCPServer; s = HomeAssistantSmartMCPServer(); s.mcp.run(transport='sse', host='0.0.0.0', port=3000)"
|
||||||
|
# For connecting to an already-running instance (host.containers.internal for container-to-host):
|
||||||
|
externalUrl: "http://host.containers.internal:8086/mcp"
|
||||||
|
envTemplate:
|
||||||
|
- name: HOMEASSISTANT_URL
|
||||||
|
description: "Home Assistant instance URL (e.g. https://ha.example.com)"
|
||||||
|
- name: HOMEASSISTANT_TOKEN
|
||||||
|
description: "Home Assistant long-lived access token"
|
||||||
|
isSecret: true
|
||||||
|
|
||||||
|
profiles:
|
||||||
|
- name: production
|
||||||
|
server: ha-mcp
|
||||||
|
envOverrides:
|
||||||
|
HOMEASSISTANT_URL: "https://your-ha-instance.example.com"
|
||||||
|
HOMEASSISTANT_TOKEN: "REDACTED-TOKEN"
|
||||||
@@ -11,6 +11,9 @@ const ServerSpecSchema = z.object({
|
|||||||
dockerImage: z.string().optional(),
|
dockerImage: z.string().optional(),
|
||||||
transport: z.enum(['STDIO', 'SSE', 'STREAMABLE_HTTP']).default('STDIO'),
|
transport: z.enum(['STDIO', 'SSE', 'STREAMABLE_HTTP']).default('STDIO'),
|
||||||
repositoryUrl: z.string().url().optional(),
|
repositoryUrl: z.string().url().optional(),
|
||||||
|
externalUrl: z.string().url().optional(),
|
||||||
|
command: z.array(z.string()).optional(),
|
||||||
|
containerPort: z.number().int().min(1).max(65535).optional(),
|
||||||
envTemplate: z.array(z.object({
|
envTemplate: z.array(z.object({
|
||||||
name: z.string(),
|
name: z.string(),
|
||||||
description: z.string().default(''),
|
description: z.string().default(''),
|
||||||
|
|||||||
@@ -0,0 +1,204 @@
|
|||||||
|
-- CreateEnum
|
||||||
|
CREATE TYPE "Role" AS ENUM ('USER', 'ADMIN');
|
||||||
|
|
||||||
|
-- CreateEnum
|
||||||
|
CREATE TYPE "Transport" AS ENUM ('STDIO', 'SSE', 'STREAMABLE_HTTP');
|
||||||
|
|
||||||
|
-- CreateEnum
|
||||||
|
CREATE TYPE "InstanceStatus" AS ENUM ('STARTING', 'RUNNING', 'STOPPING', 'STOPPED', 'ERROR');
|
||||||
|
|
||||||
|
-- CreateTable
|
||||||
|
CREATE TABLE "User" (
|
||||||
|
"id" TEXT NOT NULL,
|
||||||
|
"email" TEXT NOT NULL,
|
||||||
|
"name" TEXT,
|
||||||
|
"passwordHash" TEXT NOT NULL,
|
||||||
|
"role" "Role" NOT NULL DEFAULT 'USER',
|
||||||
|
"version" INTEGER NOT NULL DEFAULT 1,
|
||||||
|
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||||
|
"updatedAt" TIMESTAMP(3) NOT NULL,
|
||||||
|
|
||||||
|
CONSTRAINT "User_pkey" PRIMARY KEY ("id")
|
||||||
|
);
|
||||||
|
|
||||||
|
-- CreateTable
|
||||||
|
CREATE TABLE "Session" (
|
||||||
|
"id" TEXT NOT NULL,
|
||||||
|
"token" TEXT NOT NULL,
|
||||||
|
"userId" TEXT NOT NULL,
|
||||||
|
"expiresAt" TIMESTAMP(3) NOT NULL,
|
||||||
|
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||||
|
|
||||||
|
CONSTRAINT "Session_pkey" PRIMARY KEY ("id")
|
||||||
|
);
|
||||||
|
|
||||||
|
-- CreateTable
|
||||||
|
CREATE TABLE "McpServer" (
|
||||||
|
"id" TEXT NOT NULL,
|
||||||
|
"name" TEXT NOT NULL,
|
||||||
|
"description" TEXT NOT NULL DEFAULT '',
|
||||||
|
"packageName" TEXT,
|
||||||
|
"dockerImage" TEXT,
|
||||||
|
"transport" "Transport" NOT NULL DEFAULT 'STDIO',
|
||||||
|
"repositoryUrl" TEXT,
|
||||||
|
"externalUrl" TEXT,
|
||||||
|
"command" JSONB,
|
||||||
|
"containerPort" INTEGER,
|
||||||
|
"envTemplate" JSONB NOT NULL DEFAULT '[]',
|
||||||
|
"version" INTEGER NOT NULL DEFAULT 1,
|
||||||
|
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||||
|
"updatedAt" TIMESTAMP(3) NOT NULL,
|
||||||
|
|
||||||
|
CONSTRAINT "McpServer_pkey" PRIMARY KEY ("id")
|
||||||
|
);
|
||||||
|
|
||||||
|
-- CreateTable
|
||||||
|
CREATE TABLE "McpProfile" (
|
||||||
|
"id" TEXT NOT NULL,
|
||||||
|
"name" TEXT NOT NULL,
|
||||||
|
"serverId" TEXT NOT NULL,
|
||||||
|
"permissions" JSONB NOT NULL DEFAULT '[]',
|
||||||
|
"envOverrides" JSONB NOT NULL DEFAULT '{}',
|
||||||
|
"version" INTEGER NOT NULL DEFAULT 1,
|
||||||
|
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||||
|
"updatedAt" TIMESTAMP(3) NOT NULL,
|
||||||
|
|
||||||
|
CONSTRAINT "McpProfile_pkey" PRIMARY KEY ("id")
|
||||||
|
);
|
||||||
|
|
||||||
|
-- CreateTable
|
||||||
|
CREATE TABLE "Project" (
|
||||||
|
"id" TEXT NOT NULL,
|
||||||
|
"name" TEXT NOT NULL,
|
||||||
|
"description" TEXT NOT NULL DEFAULT '',
|
||||||
|
"ownerId" TEXT NOT NULL,
|
||||||
|
"version" INTEGER NOT NULL DEFAULT 1,
|
||||||
|
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||||
|
"updatedAt" TIMESTAMP(3) NOT NULL,
|
||||||
|
|
||||||
|
CONSTRAINT "Project_pkey" PRIMARY KEY ("id")
|
||||||
|
);
|
||||||
|
|
||||||
|
-- CreateTable
|
||||||
|
CREATE TABLE "ProjectMcpProfile" (
|
||||||
|
"id" TEXT NOT NULL,
|
||||||
|
"projectId" TEXT NOT NULL,
|
||||||
|
"profileId" TEXT NOT NULL,
|
||||||
|
|
||||||
|
CONSTRAINT "ProjectMcpProfile_pkey" PRIMARY KEY ("id")
|
||||||
|
);
|
||||||
|
|
||||||
|
-- CreateTable
|
||||||
|
CREATE TABLE "McpInstance" (
|
||||||
|
"id" TEXT NOT NULL,
|
||||||
|
"serverId" TEXT NOT NULL,
|
||||||
|
"containerId" TEXT,
|
||||||
|
"status" "InstanceStatus" NOT NULL DEFAULT 'STOPPED',
|
||||||
|
"port" INTEGER,
|
||||||
|
"metadata" JSONB NOT NULL DEFAULT '{}',
|
||||||
|
"version" INTEGER NOT NULL DEFAULT 1,
|
||||||
|
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||||
|
"updatedAt" TIMESTAMP(3) NOT NULL,
|
||||||
|
|
||||||
|
CONSTRAINT "McpInstance_pkey" PRIMARY KEY ("id")
|
||||||
|
);
|
||||||
|
|
||||||
|
-- CreateTable
|
||||||
|
CREATE TABLE "AuditLog" (
|
||||||
|
"id" TEXT NOT NULL,
|
||||||
|
"userId" TEXT NOT NULL,
|
||||||
|
"action" TEXT NOT NULL,
|
||||||
|
"resource" TEXT NOT NULL,
|
||||||
|
"resourceId" TEXT,
|
||||||
|
"details" JSONB NOT NULL DEFAULT '{}',
|
||||||
|
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||||
|
|
||||||
|
CONSTRAINT "AuditLog_pkey" PRIMARY KEY ("id")
|
||||||
|
);
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE UNIQUE INDEX "User_email_key" ON "User"("email");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE INDEX "User_email_idx" ON "User"("email");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE UNIQUE INDEX "Session_token_key" ON "Session"("token");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE INDEX "Session_token_idx" ON "Session"("token");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE INDEX "Session_userId_idx" ON "Session"("userId");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE INDEX "Session_expiresAt_idx" ON "Session"("expiresAt");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE UNIQUE INDEX "McpServer_name_key" ON "McpServer"("name");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE INDEX "McpServer_name_idx" ON "McpServer"("name");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE INDEX "McpProfile_serverId_idx" ON "McpProfile"("serverId");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE UNIQUE INDEX "McpProfile_name_serverId_key" ON "McpProfile"("name", "serverId");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE UNIQUE INDEX "Project_name_key" ON "Project"("name");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE INDEX "Project_name_idx" ON "Project"("name");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE INDEX "Project_ownerId_idx" ON "Project"("ownerId");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE INDEX "ProjectMcpProfile_projectId_idx" ON "ProjectMcpProfile"("projectId");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE INDEX "ProjectMcpProfile_profileId_idx" ON "ProjectMcpProfile"("profileId");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE UNIQUE INDEX "ProjectMcpProfile_projectId_profileId_key" ON "ProjectMcpProfile"("projectId", "profileId");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE INDEX "McpInstance_serverId_idx" ON "McpInstance"("serverId");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE INDEX "McpInstance_status_idx" ON "McpInstance"("status");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE INDEX "AuditLog_userId_idx" ON "AuditLog"("userId");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE INDEX "AuditLog_action_idx" ON "AuditLog"("action");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE INDEX "AuditLog_resource_idx" ON "AuditLog"("resource");
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE INDEX "AuditLog_createdAt_idx" ON "AuditLog"("createdAt");
|
||||||
|
|
||||||
|
-- AddForeignKey
|
||||||
|
ALTER TABLE "Session" ADD CONSTRAINT "Session_userId_fkey" FOREIGN KEY ("userId") REFERENCES "User"("id") ON DELETE CASCADE ON UPDATE CASCADE;
|
||||||
|
|
||||||
|
-- AddForeignKey
|
||||||
|
ALTER TABLE "McpProfile" ADD CONSTRAINT "McpProfile_serverId_fkey" FOREIGN KEY ("serverId") REFERENCES "McpServer"("id") ON DELETE CASCADE ON UPDATE CASCADE;
|
||||||
|
|
||||||
|
-- AddForeignKey
|
||||||
|
ALTER TABLE "Project" ADD CONSTRAINT "Project_ownerId_fkey" FOREIGN KEY ("ownerId") REFERENCES "User"("id") ON DELETE CASCADE ON UPDATE CASCADE;
|
||||||
|
|
||||||
|
-- AddForeignKey
|
||||||
|
ALTER TABLE "ProjectMcpProfile" ADD CONSTRAINT "ProjectMcpProfile_projectId_fkey" FOREIGN KEY ("projectId") REFERENCES "Project"("id") ON DELETE CASCADE ON UPDATE CASCADE;
|
||||||
|
|
||||||
|
-- AddForeignKey
|
||||||
|
ALTER TABLE "ProjectMcpProfile" ADD CONSTRAINT "ProjectMcpProfile_profileId_fkey" FOREIGN KEY ("profileId") REFERENCES "McpProfile"("id") ON DELETE CASCADE ON UPDATE CASCADE;
|
||||||
|
|
||||||
|
-- AddForeignKey
|
||||||
|
ALTER TABLE "McpInstance" ADD CONSTRAINT "McpInstance_serverId_fkey" FOREIGN KEY ("serverId") REFERENCES "McpServer"("id") ON DELETE CASCADE ON UPDATE CASCADE;
|
||||||
|
|
||||||
|
-- AddForeignKey
|
||||||
|
ALTER TABLE "AuditLog" ADD CONSTRAINT "AuditLog_userId_fkey" FOREIGN KEY ("userId") REFERENCES "User"("id") ON DELETE CASCADE ON UPDATE CASCADE;
|
||||||
3
src/db/prisma/migrations/migration_lock.toml
Normal file
3
src/db/prisma/migrations/migration_lock.toml
Normal file
@@ -0,0 +1,3 @@
|
|||||||
|
# Please do not edit this file manually
|
||||||
|
# It should be added in your version-control system (e.g., Git)
|
||||||
|
provider = "postgresql"
|
||||||
@@ -57,6 +57,9 @@ model McpServer {
|
|||||||
dockerImage String?
|
dockerImage String?
|
||||||
transport Transport @default(STDIO)
|
transport Transport @default(STDIO)
|
||||||
repositoryUrl String?
|
repositoryUrl String?
|
||||||
|
externalUrl String?
|
||||||
|
command Json?
|
||||||
|
containerPort Int?
|
||||||
envTemplate Json @default("[]")
|
envTemplate Json @default("[]")
|
||||||
version Int @default(1)
|
version Int @default(1)
|
||||||
createdAt DateTime @default(now())
|
createdAt DateTime @default(now())
|
||||||
|
|||||||
@@ -69,7 +69,7 @@ async function main(): Promise<void> {
|
|||||||
const backupService = new BackupService(serverRepo, profileRepo, projectRepo);
|
const backupService = new BackupService(serverRepo, profileRepo, projectRepo);
|
||||||
const restoreService = new RestoreService(serverRepo, profileRepo, projectRepo);
|
const restoreService = new RestoreService(serverRepo, profileRepo, projectRepo);
|
||||||
const authService = new AuthService(prisma);
|
const authService = new AuthService(prisma);
|
||||||
const mcpProxyService = new McpProxyService(instanceRepo);
|
const mcpProxyService = new McpProxyService(instanceRepo, serverRepo);
|
||||||
|
|
||||||
// Server
|
// Server
|
||||||
const app = await createServer(config, {
|
const app = await createServer(config, {
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
import type { PrismaClient, McpServer } from '@prisma/client';
|
import { type PrismaClient, type McpServer, Prisma } from '@prisma/client';
|
||||||
import type { IMcpServerRepository } from './interfaces.js';
|
import type { IMcpServerRepository } from './interfaces.js';
|
||||||
import type { CreateMcpServerInput, UpdateMcpServerInput } from '../validation/mcp-server.schema.js';
|
import type { CreateMcpServerInput, UpdateMcpServerInput } from '../validation/mcp-server.schema.js';
|
||||||
|
|
||||||
@@ -26,6 +26,9 @@ export class McpServerRepository implements IMcpServerRepository {
|
|||||||
dockerImage: data.dockerImage ?? null,
|
dockerImage: data.dockerImage ?? null,
|
||||||
transport: data.transport,
|
transport: data.transport,
|
||||||
repositoryUrl: data.repositoryUrl ?? null,
|
repositoryUrl: data.repositoryUrl ?? null,
|
||||||
|
externalUrl: data.externalUrl ?? null,
|
||||||
|
command: data.command ?? Prisma.DbNull,
|
||||||
|
containerPort: data.containerPort ?? null,
|
||||||
envTemplate: data.envTemplate,
|
envTemplate: data.envTemplate,
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
@@ -38,6 +41,9 @@ export class McpServerRepository implements IMcpServerRepository {
|
|||||||
if (data.dockerImage !== undefined) updateData['dockerImage'] = data.dockerImage;
|
if (data.dockerImage !== undefined) updateData['dockerImage'] = data.dockerImage;
|
||||||
if (data.transport !== undefined) updateData['transport'] = data.transport;
|
if (data.transport !== undefined) updateData['transport'] = data.transport;
|
||||||
if (data.repositoryUrl !== undefined) updateData['repositoryUrl'] = data.repositoryUrl;
|
if (data.repositoryUrl !== undefined) updateData['repositoryUrl'] = data.repositoryUrl;
|
||||||
|
if (data.externalUrl !== undefined) updateData['externalUrl'] = data.externalUrl;
|
||||||
|
if (data.command !== undefined) updateData['command'] = data.command;
|
||||||
|
if (data.containerPort !== undefined) updateData['containerPort'] = data.containerPort;
|
||||||
if (data.envTemplate !== undefined) updateData['envTemplate'] = data.envTemplate;
|
if (data.envTemplate !== undefined) updateData['envTemplate'] = data.envTemplate;
|
||||||
|
|
||||||
return this.prisma.mcpServer.update({ where: { id }, data: updateData });
|
return this.prisma.mcpServer.update({ where: { id }, data: updateData });
|
||||||
|
|||||||
@@ -74,7 +74,7 @@ export class DockerContainerManager implements McpOrchestrator {
|
|||||||
? Object.entries(spec.env).map(([k, v]) => `${k}=${v}`)
|
? Object.entries(spec.env).map(([k, v]) => `${k}=${v}`)
|
||||||
: undefined;
|
: undefined;
|
||||||
|
|
||||||
const container = await this.docker.createContainer({
|
const createOpts: Docker.ContainerCreateOptions = {
|
||||||
Image: spec.image,
|
Image: spec.image,
|
||||||
name: spec.name,
|
name: spec.name,
|
||||||
Env: envArr,
|
Env: envArr,
|
||||||
@@ -86,7 +86,12 @@ export class DockerContainerManager implements McpOrchestrator {
|
|||||||
NanoCpus: nanoCpus,
|
NanoCpus: nanoCpus,
|
||||||
NetworkMode: spec.network ?? 'bridge',
|
NetworkMode: spec.network ?? 'bridge',
|
||||||
},
|
},
|
||||||
});
|
};
|
||||||
|
if (spec.command) {
|
||||||
|
createOpts.Cmd = spec.command;
|
||||||
|
}
|
||||||
|
|
||||||
|
const container = await this.docker.createContainer(createOpts);
|
||||||
|
|
||||||
await container.start();
|
await container.start();
|
||||||
|
|
||||||
|
|||||||
@@ -32,6 +32,15 @@ export class InstanceService {
|
|||||||
const server = await this.serverRepo.findById(serverId);
|
const server = await this.serverRepo.findById(serverId);
|
||||||
if (!server) throw new NotFoundError(`McpServer '${serverId}' not found`);
|
if (!server) throw new NotFoundError(`McpServer '${serverId}' not found`);
|
||||||
|
|
||||||
|
// External servers don't need container management
|
||||||
|
if (server.externalUrl) {
|
||||||
|
return this.instanceRepo.create({
|
||||||
|
serverId,
|
||||||
|
status: 'RUNNING',
|
||||||
|
metadata: { external: true, url: server.externalUrl },
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
const image = server.dockerImage ?? server.packageName ?? server.name;
|
const image = server.dockerImage ?? server.packageName ?? server.name;
|
||||||
|
|
||||||
// Create DB record first in STARTING state
|
// Create DB record first in STARTING state
|
||||||
@@ -51,7 +60,11 @@ export class InstanceService {
|
|||||||
},
|
},
|
||||||
};
|
};
|
||||||
if (server.transport === 'SSE' || server.transport === 'STREAMABLE_HTTP') {
|
if (server.transport === 'SSE' || server.transport === 'STREAMABLE_HTTP') {
|
||||||
spec.containerPort = 3000;
|
spec.containerPort = server.containerPort ?? 3000;
|
||||||
|
}
|
||||||
|
const command = server.command as string[] | null;
|
||||||
|
if (command) {
|
||||||
|
spec.command = command;
|
||||||
}
|
}
|
||||||
if (opts?.env) {
|
if (opts?.env) {
|
||||||
spec.env = opts.env;
|
spec.env = opts.env;
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
import type { McpInstance } from '@prisma/client';
|
import type { McpInstance } from '@prisma/client';
|
||||||
import type { IMcpInstanceRepository } from '../repositories/interfaces.js';
|
import type { IMcpInstanceRepository, IMcpServerRepository } from '../repositories/interfaces.js';
|
||||||
import { NotFoundError } from './mcp-server.service.js';
|
import { NotFoundError } from './mcp-server.service.js';
|
||||||
import { InvalidStateError } from './instance.service.js';
|
import { InvalidStateError } from './instance.service.js';
|
||||||
|
|
||||||
@@ -16,11 +16,39 @@ export interface McpProxyResponse {
|
|||||||
error?: { code: number; message: string; data?: unknown };
|
error?: { code: number; message: string; data?: unknown };
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Parses a streamable-http SSE response body to extract the JSON-RPC payload.
|
||||||
|
* Streamable-http returns `event: message\ndata: {...}\n\n` format.
|
||||||
|
*/
|
||||||
|
function parseStreamableResponse(body: string): McpProxyResponse {
|
||||||
|
for (const line of body.split('\n')) {
|
||||||
|
const trimmed = line.trim();
|
||||||
|
if (trimmed.startsWith('data: ')) {
|
||||||
|
return JSON.parse(trimmed.slice(6)) as McpProxyResponse;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// If body is plain JSON (no SSE framing), parse directly
|
||||||
|
return JSON.parse(body) as McpProxyResponse;
|
||||||
|
}
|
||||||
|
|
||||||
export class McpProxyService {
|
export class McpProxyService {
|
||||||
constructor(private readonly instanceRepo: IMcpInstanceRepository) {}
|
/** Session IDs per server for streamable-http protocol */
|
||||||
|
private sessions = new Map<string, string>();
|
||||||
|
|
||||||
|
constructor(
|
||||||
|
private readonly instanceRepo: IMcpInstanceRepository,
|
||||||
|
private readonly serverRepo: IMcpServerRepository,
|
||||||
|
) {}
|
||||||
|
|
||||||
async execute(request: McpProxyRequest): Promise<McpProxyResponse> {
|
async execute(request: McpProxyRequest): Promise<McpProxyResponse> {
|
||||||
// Find a running instance for this server
|
const server = await this.serverRepo.findById(request.serverId);
|
||||||
|
|
||||||
|
// External server: proxy directly to externalUrl
|
||||||
|
if (server?.externalUrl) {
|
||||||
|
return this.sendToExternal(server.id, server.externalUrl, request.method, request.params);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Managed server: find running instance
|
||||||
const instances = await this.instanceRepo.findAll(request.serverId);
|
const instances = await this.instanceRepo.findAll(request.serverId);
|
||||||
const running = instances.find((i) => i.status === 'RUNNING');
|
const running = instances.find((i) => i.status === 'RUNNING');
|
||||||
|
|
||||||
@@ -37,6 +65,116 @@ export class McpProxyService {
|
|||||||
return this.sendJsonRpc(running, request.method, request.params);
|
return this.sendJsonRpc(running, request.method, request.params);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Send a JSON-RPC request to an external MCP server.
|
||||||
|
* Handles streamable-http protocol (session management + SSE response parsing).
|
||||||
|
*/
|
||||||
|
private async sendToExternal(
|
||||||
|
serverId: string,
|
||||||
|
url: string,
|
||||||
|
method: string,
|
||||||
|
params?: Record<string, unknown>,
|
||||||
|
): Promise<McpProxyResponse> {
|
||||||
|
// Ensure we have a session (initialize on first call)
|
||||||
|
if (!this.sessions.has(serverId)) {
|
||||||
|
await this.initSession(serverId, url);
|
||||||
|
}
|
||||||
|
|
||||||
|
const sessionId = this.sessions.get(serverId);
|
||||||
|
|
||||||
|
const body: Record<string, unknown> = {
|
||||||
|
jsonrpc: '2.0',
|
||||||
|
id: 1,
|
||||||
|
method,
|
||||||
|
};
|
||||||
|
if (params !== undefined) {
|
||||||
|
body.params = params;
|
||||||
|
}
|
||||||
|
|
||||||
|
const headers: Record<string, string> = {
|
||||||
|
'Content-Type': 'application/json',
|
||||||
|
'Accept': 'application/json, text/event-stream',
|
||||||
|
};
|
||||||
|
if (sessionId) {
|
||||||
|
headers['Mcp-Session-Id'] = sessionId;
|
||||||
|
}
|
||||||
|
|
||||||
|
const response = await fetch(url, {
|
||||||
|
method: 'POST',
|
||||||
|
headers,
|
||||||
|
body: JSON.stringify(body),
|
||||||
|
});
|
||||||
|
|
||||||
|
if (!response.ok) {
|
||||||
|
// Session expired? Clear and retry once
|
||||||
|
if (response.status === 400 || response.status === 404) {
|
||||||
|
this.sessions.delete(serverId);
|
||||||
|
return this.sendToExternal(serverId, url, method, params);
|
||||||
|
}
|
||||||
|
return {
|
||||||
|
jsonrpc: '2.0',
|
||||||
|
id: 1,
|
||||||
|
error: {
|
||||||
|
code: -32000,
|
||||||
|
message: `External MCP server returned HTTP ${response.status}: ${response.statusText}`,
|
||||||
|
},
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
const text = await response.text();
|
||||||
|
return parseStreamableResponse(text);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Initialize a streamable-http session with an external server.
|
||||||
|
* Sends `initialize` and `notifications/initialized`, caches the session ID.
|
||||||
|
*/
|
||||||
|
private async initSession(serverId: string, url: string): Promise<void> {
|
||||||
|
const initBody = {
|
||||||
|
jsonrpc: '2.0',
|
||||||
|
id: 1,
|
||||||
|
method: 'initialize',
|
||||||
|
params: {
|
||||||
|
protocolVersion: '2025-03-26',
|
||||||
|
capabilities: {},
|
||||||
|
clientInfo: { name: 'mcpctl', version: '0.1.0' },
|
||||||
|
},
|
||||||
|
};
|
||||||
|
|
||||||
|
const response = await fetch(url, {
|
||||||
|
method: 'POST',
|
||||||
|
headers: {
|
||||||
|
'Content-Type': 'application/json',
|
||||||
|
'Accept': 'application/json, text/event-stream',
|
||||||
|
},
|
||||||
|
body: JSON.stringify(initBody),
|
||||||
|
});
|
||||||
|
|
||||||
|
if (!response.ok) {
|
||||||
|
throw new Error(`Failed to initialize session: HTTP ${response.status}`);
|
||||||
|
}
|
||||||
|
|
||||||
|
const sessionId = response.headers.get('mcp-session-id');
|
||||||
|
if (sessionId) {
|
||||||
|
this.sessions.set(serverId, sessionId);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Send notifications/initialized
|
||||||
|
const headers: Record<string, string> = {
|
||||||
|
'Content-Type': 'application/json',
|
||||||
|
'Accept': 'application/json, text/event-stream',
|
||||||
|
};
|
||||||
|
if (sessionId) {
|
||||||
|
headers['Mcp-Session-Id'] = sessionId;
|
||||||
|
}
|
||||||
|
|
||||||
|
await fetch(url, {
|
||||||
|
method: 'POST',
|
||||||
|
headers,
|
||||||
|
body: JSON.stringify({ jsonrpc: '2.0', method: 'notifications/initialized' }),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
private async sendJsonRpc(
|
private async sendJsonRpc(
|
||||||
instance: McpInstance,
|
instance: McpInstance,
|
||||||
method: string,
|
method: string,
|
||||||
|
|||||||
@@ -7,6 +7,8 @@ export interface ContainerSpec {
|
|||||||
image: string;
|
image: string;
|
||||||
/** Human-readable name (used as container name prefix) */
|
/** Human-readable name (used as container name prefix) */
|
||||||
name: string;
|
name: string;
|
||||||
|
/** Custom command to run (overrides image CMD) */
|
||||||
|
command?: string[];
|
||||||
/** Environment variables */
|
/** Environment variables */
|
||||||
env?: Record<string, string>;
|
env?: Record<string, string>;
|
||||||
/** Host port to bind (null = auto-assign) */
|
/** Host port to bind (null = auto-assign) */
|
||||||
|
|||||||
@@ -14,6 +14,9 @@ export const CreateMcpServerSchema = z.object({
|
|||||||
dockerImage: z.string().max(200).optional(),
|
dockerImage: z.string().max(200).optional(),
|
||||||
transport: z.enum(['STDIO', 'SSE', 'STREAMABLE_HTTP']).default('STDIO'),
|
transport: z.enum(['STDIO', 'SSE', 'STREAMABLE_HTTP']).default('STDIO'),
|
||||||
repositoryUrl: z.string().url().optional(),
|
repositoryUrl: z.string().url().optional(),
|
||||||
|
externalUrl: z.string().url().optional(),
|
||||||
|
command: z.array(z.string()).optional(),
|
||||||
|
containerPort: z.number().int().min(1).max(65535).optional(),
|
||||||
envTemplate: z.array(EnvTemplateEntrySchema).default([]),
|
envTemplate: z.array(EnvTemplateEntrySchema).default([]),
|
||||||
});
|
});
|
||||||
|
|
||||||
@@ -23,6 +26,9 @@ export const UpdateMcpServerSchema = z.object({
|
|||||||
dockerImage: z.string().max(200).nullable().optional(),
|
dockerImage: z.string().max(200).nullable().optional(),
|
||||||
transport: z.enum(['STDIO', 'SSE', 'STREAMABLE_HTTP']).optional(),
|
transport: z.enum(['STDIO', 'SSE', 'STREAMABLE_HTTP']).optional(),
|
||||||
repositoryUrl: z.string().url().nullable().optional(),
|
repositoryUrl: z.string().url().nullable().optional(),
|
||||||
|
externalUrl: z.string().url().nullable().optional(),
|
||||||
|
command: z.array(z.string()).nullable().optional(),
|
||||||
|
containerPort: z.number().int().min(1).max(65535).nullable().optional(),
|
||||||
envTemplate: z.array(EnvTemplateEntrySchema).optional(),
|
envTemplate: z.array(EnvTemplateEntrySchema).optional(),
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user