History

add unified source connect and discover API endpoints

Introduce POST /sources/connect that handles MCP, OpenAPI, and GraphQL connection in a single endpoint (replacing the old MCP-only connect). Add POST /sources/discover for URL-based source type detection. Update sources-operations with auto-probe logic for draft sources.

Rhys Sullivan committed 8 months ago

Commit af725aa1 parent ff7c45f3 files · +266 −90Browse files

Unified · Split

3 files
  1. packages/control-plane/src/api/sources/api.ts +105 −38
  2. packages/control-plane/src/api/sources/http.ts +119 −12
  3. packages/control-plane/src/runtime/sources-operations.ts +42 −40
packages/control-plane/src/api/sources/api.ts+105−38
@@ -1,14 +1,17 @@
11 import { HttpApiEndpoint, HttpApiGroup, HttpApiSchema } from "@effect/platform";
22 import {
33 ExecutionInteractionIdSchema,
4+ SecretRefSchema,
45 SourceAuthSchema,
56 SourceAuthSessionIdSchema,
7+ SourceDiscoveryResultSchema,
68 SourceIdSchema,
79 SourceInspectionDiscoverPayloadSchema,
810 SourceInspectionDiscoverResultSchema,
911 SourceInspectionSchema,
1012 SourceInspectionToolDetailSchema,
1113 SourceKindSchema,
14+ SourceProbeAuthSchema,
1215 SourceSchema,
1316 SourceStatusSchema,
1417 SourceTransportSchema,
@@ -75,34 +78,6 @@ export const UpdateSourcePayloadSchema = Schema.Struct({ · 60 unmodified lines
7578
7679 export type UpdateSourcePayload = typeof UpdateSourcePayloadSchema.Type;
7780
78−export const ConnectMcpSourcePayloadSchema = Schema.Struct({
79− sourceId: Schema.optional(SourceIdSchema),
80− name: Schema.optional(Schema.NullOr(Schema.String)),
81− endpoint: TrimmedNonEmptyStringSchema,
82− namespace: Schema.optional(Schema.NullOr(Schema.String)),
83− enabled: Schema.optional(Schema.Boolean),
84− transport: Schema.optional(SourceTransportSchema),
85− queryParams: Schema.optional(Schema.NullOr(StringMapSchema)),
86− headers: Schema.optional(Schema.NullOr(StringMapSchema)),
87−});
88−
89−export type ConnectMcpSourcePayload = typeof ConnectMcpSourcePayloadSchema.Type;
90−
91−export const ConnectMcpSourceResultSchema = Schema.Union(
92− Schema.Struct({
93− kind: Schema.Literal("connected"),
94− source: SourceSchema,
95− }),
96− Schema.Struct({
97− kind: Schema.Literal("oauth_required"),
98− source: SourceSchema,
99− sessionId: SourceAuthSessionIdSchema,
100− authorizationUrl: Schema.String,
101− }),
102−);
103−
104−export type ConnectMcpSourceResult = typeof ConnectMcpSourceResultSchema.Type;
105−
10681 const workspaceIdParam = HttpApiSchema.param("workspaceId", WorkspaceIdSchema);
10782 const sourceIdParam = HttpApiSchema.param("sourceId", SourceIdSchema);
10883 const toolPathParam = HttpApiSchema.param("toolPath", Schema.String);
@@ -123,11 +98,113 @@ const CredentialOauthCompleteUrlParamsSchema = Schema.Struct({ · 14 unmodified lines
12398 error_description: Schema.optional(Schema.String),
12499 });
125100
101+const DiscoverSourcePayloadSchema = Schema.Struct({
102+ url: TrimmedNonEmptyStringSchema,
103+ probeAuth: Schema.optional(SourceProbeAuthSchema),
104+});
105+
106+export type DiscoverSourcePayload = typeof DiscoverSourcePayloadSchema.Type;
107+
108+const ConnectBearerAuthSchema = Schema.Struct({
109+ kind: Schema.Literal("bearer"),
110+ headerName: Schema.optional(Schema.NullOr(Schema.String)),
111+ prefix: Schema.optional(Schema.NullOr(Schema.String)),
112+ token: Schema.optional(Schema.NullOr(Schema.String)),
113+ tokenRef: Schema.optional(Schema.NullOr(SecretRefSchema)),
114+});
115+
116+const ConnectOAuth2AuthSchema = Schema.Struct({
117+ kind: Schema.Literal("oauth2"),
118+ headerName: Schema.optional(Schema.NullOr(Schema.String)),
119+ prefix: Schema.optional(Schema.NullOr(Schema.String)),
120+ accessToken: Schema.optional(Schema.NullOr(Schema.String)),
121+ accessTokenRef: Schema.optional(Schema.NullOr(SecretRefSchema)),
122+ refreshToken: Schema.optional(Schema.NullOr(Schema.String)),
123+ refreshTokenRef: Schema.optional(Schema.NullOr(SecretRefSchema)),
124+});
125+
126+const ConnectHttpAuthSchema = Schema.Union(
127+ Schema.Struct({ kind: Schema.Literal("none") }),
128+ ConnectBearerAuthSchema,
129+ ConnectOAuth2AuthSchema,
130+);
131+
132+const ConnectSourcePayloadSchema = Schema.Union(
133+ Schema.Struct({
134+ kind: Schema.Literal("openapi"),
135+ endpoint: TrimmedNonEmptyStringSchema,
136+ specUrl: TrimmedNonEmptyStringSchema,
137+ name: Schema.optional(Schema.NullOr(Schema.String)),
138+ namespace: Schema.optional(Schema.NullOr(Schema.String)),
139+ auth: Schema.optional(ConnectHttpAuthSchema),
140+ }),
141+ Schema.Struct({
142+ kind: Schema.Literal("graphql"),
143+ endpoint: TrimmedNonEmptyStringSchema,
144+ name: Schema.optional(Schema.NullOr(Schema.String)),
145+ namespace: Schema.optional(Schema.NullOr(Schema.String)),
146+ auth: Schema.optional(ConnectHttpAuthSchema),
147+ }),
148+ Schema.Struct({
149+ kind: Schema.Literal("mcp"),
150+ endpoint: TrimmedNonEmptyStringSchema,
151+ name: Schema.optional(Schema.NullOr(Schema.String)),
152+ namespace: Schema.optional(Schema.NullOr(Schema.String)),
153+ transport: Schema.optional(Schema.NullOr(SourceTransportSchema)),
154+ queryParams: Schema.optional(Schema.NullOr(StringMapSchema)),
155+ headers: Schema.optional(Schema.NullOr(StringMapSchema)),
156+ }),
157+);
158+
159+export type ConnectSourcePayload = typeof ConnectSourcePayloadSchema.Type;
160+
161+const ConnectSourceResultSchema = Schema.Union(
162+ Schema.Struct({
163+ kind: Schema.Literal("connected"),
164+ source: SourceSchema,
165+ }),
166+ Schema.Struct({
167+ kind: Schema.Literal("credential_required"),
168+ source: SourceSchema,
169+ }),
170+ Schema.Struct({
171+ kind: Schema.Literal("oauth_required"),
172+ source: SourceSchema,
173+ sessionId: SourceAuthSessionIdSchema,
174+ authorizationUrl: Schema.String,
175+ }),
176+);
177+
178+export type ConnectSourceResult = typeof ConnectSourceResultSchema.Type;
179+
180+export {
181+ ConnectSourcePayloadSchema,
182+ ConnectSourceResultSchema,
183+ DiscoverSourcePayloadSchema,
184+};
185+
126186 const HtmlSchema = HttpApiSchema.Text({
127187 contentType: "text/html",
128188 });
129189
130190 export class SourcesApi extends HttpApiGroup.make("sources")
191+ .add(
192+ HttpApiEndpoint.post("discover")`/sources/discover`
193+ .setPayload(DiscoverSourcePayloadSchema)
194+ .addSuccess(SourceDiscoveryResultSchema)
195+ .addError(ControlPlaneBadRequestError)
196+ .addError(ControlPlaneUnauthorizedError)
197+ .addError(ControlPlaneForbiddenError),
198+ )
199+ .add(
200+ HttpApiEndpoint.post("connect")`/workspaces/${workspaceIdParam}/sources/connect`
201+ .setPayload(ConnectSourcePayloadSchema)
202+ .addSuccess(ConnectSourceResultSchema)
203+ .addError(ControlPlaneBadRequestError)
204+ .addError(ControlPlaneUnauthorizedError)
205+ .addError(ControlPlaneForbiddenError)
206+ .addError(ControlPlaneStorageError),
207+ )
131208 .add(
132209 HttpApiEndpoint.get("list")`/workspaces/${workspaceIdParam}/sources`
133210 .addSuccess(Schema.Array(SourceSchema))
@@ -164,16 +241,6 @@ export class SourcesApi extends HttpApiGroup.make("sources") · 30 unmodified lines
164241 .addError(ControlPlaneNotFoundError)
165242 .addError(ControlPlaneStorageError),
166243 )
167− .add(
168− HttpApiEndpoint.post("connectMcp")`/workspaces/${workspaceIdParam}/sources/mcp/oauth/start`
169− .setPayload(ConnectMcpSourcePayloadSchema)
170− .addSuccess(ConnectMcpSourceResultSchema)
171− .addError(ControlPlaneBadRequestError)
172− .addError(ControlPlaneUnauthorizedError)
173− .addError(ControlPlaneForbiddenError)
174− .addError(ControlPlaneNotFoundError)
175− .addError(ControlPlaneStorageError),
176− )
177244 .add(
178245 HttpApiEndpoint.del("remove")`/workspaces/${workspaceIdParam}/sources/${sourceIdParam}`
179246 .addSuccess(Schema.Struct({ removed: Schema.Boolean }))
packages/control-plane/src/api/sources/http.ts+119−12
@@ -1,5 +1,6 @@
11 import {
22 HttpApiBuilder,
3+ HttpServerRequest,
34 HttpServerResponse,
45 } from "@effect/platform";
56 import type {
@@ -11,7 +12,6 @@ import * as Effect from "effect/Effect"; · 5 unmodified lines
1112
1213 import { requirePermission, withPolicy } from "#domain";
1314 import {
14− connectMcpSource,
1515 createSource,
1616 getSource,
1717 listSources,
@@ -28,6 +28,8 @@ import { · 10 unmodified lines
2828 getSourceCredentialInteraction,
2929 submitSourceCredentialInteraction,
3030 } from "../../runtime/local-operations";
31+import { discoverSource } from "../../runtime/source-discovery";
32+import { RuntimeSourceAuthServiceTag } from "../../runtime/source-auth-service";
3133
3234 import {
3335 ControlPlaneBadRequestError,
@@ -35,7 +37,7 @@ import { · 1 unmodified line
3537 ControlPlaneStorageError,
3638 } from "../errors";
3739 import { ControlPlaneApi } from "../api";
38−import { withWorkspaceRequestActor } from "../http-auth";
40+import { withRequestActor, withWorkspaceRequestActor } from "../http-auth";
3941
4042 const requireReadSources = (workspaceId: WorkspaceId) =>
4143 requirePermission({
@@ -49,6 +51,50 @@ const requireWriteSources = (workspaceId: WorkspaceId) => · 7 unmodified lines
4951 workspaceId,
5052 });
5153
54+const readHeader = (headers: unknown, name: string): string | null => {
55+ if (headers == null || typeof headers !== "object") {
56+ return null;
57+ }
58+
59+ const record = headers as Record<string, unknown>;
60+ const exact = record[name];
61+ if (typeof exact === "string" && exact.length > 0) {
62+ return exact;
63+ }
64+
65+ const lower = record[name.toLowerCase()];
66+ if (typeof lower === "string" && lower.length > 0) {
67+ return lower;
68+ }
69+
70+ return null;
71+};
72+
73+const resolveRequestOrigin = (request: { url: string; headers: unknown }): string | null => {
74+ try {
75+ return new URL(request.url).origin;
76+ } catch {
77+ const forwardedHost = readHeader(request.headers, "x-forwarded-host");
78+ const host = forwardedHost ?? readHeader(request.headers, "host");
79+ if (!host) {
80+ return null;
81+ }
82+
83+ const forwardedProto = readHeader(request.headers, "x-forwarded-proto");
84+ const protocol = forwardedProto && forwardedProto.length > 0 ? forwardedProto : "http";
85+ return `${protocol}://${host}`;
86+ }
87+};
88+
89+const toBadRequestError = (operation: string, cause: unknown) => {
90+ const message = cause instanceof Error ? cause.message : String(cause);
91+ return new ControlPlaneBadRequestError({
92+ operation,
93+ message,
94+ details: message,
95+ });
96+};
97+
5298 const escapeHtml = (value: string): string =>
5399 value
54100 .replaceAll("&", "&amp;")
@@ -616,6 +662,77 @@ export const ControlPlaneSourcesLive = HttpApiBuilder.group( · 561 unmodified lines
616662 "sources",
617663 (handlers) =>
618664 handlers
665+ .handle("discover", ({ payload }) =>
666+ withRequestActor("sources.discover", () =>
667+ discoverSource({
668+ url: payload.url,
669+ probeAuth: payload.probeAuth,
670+ }).pipe(
671+ Effect.catchAll((cause) =>
672+ Effect.fail(toBadRequestError("sources.discover", cause)),
673+ ),
674+ ),
675+ ),
676+ )
677+ .handle("connect", ({ path, payload }) =>
678+ withWorkspaceRequestActor("sources.connect", path.workspaceId, () =>
679+ withPolicy(requireWriteSources(path.workspaceId))(
680+ Effect.gen(function* () {
681+ const request = yield* HttpServerRequest.HttpServerRequest;
682+ const sourceAuthService = yield* RuntimeSourceAuthServiceTag;
683+ const baseUrl = resolveRequestOrigin(request);
684+
685+ if (payload.kind === "mcp") {
686+ return yield* sourceAuthService.connectMcpSource({
687+ workspaceId: path.workspaceId,
688+ endpoint: payload.endpoint,
689+ name: payload.name,
690+ namespace: payload.namespace,
691+ transport: payload.transport,
692+ queryParams: payload.queryParams,
693+ headers: payload.headers,
694+ baseUrl,
695+ });
696+ }
697+
698+ if (payload.kind === "openapi") {
699+ return yield* sourceAuthService.addExecutorSource(
700+ {
701+ workspaceId: path.workspaceId,
702+ executionId: null,
703+ interactionId: null,
704+ kind: "openapi",
705+ endpoint: payload.endpoint,
706+ specUrl: payload.specUrl,
707+ name: payload.name,
708+ namespace: payload.namespace,
709+ auth: payload.auth,
710+ },
711+ { baseUrl },
712+ );
713+ }
714+
715+ return yield* sourceAuthService.addExecutorSource(
716+ {
717+ workspaceId: path.workspaceId,
718+ executionId: null,
719+ interactionId: null,
720+ kind: "graphql",
721+ endpoint: payload.endpoint,
722+ name: payload.name,
723+ namespace: payload.namespace,
724+ auth: payload.auth,
725+ },
726+ { baseUrl },
727+ );
728+ }).pipe(
729+ Effect.catchAll((cause) =>
730+ Effect.fail(toBadRequestError("sources.connect", cause)),
731+ ),
732+ ),
733+ ),
734+ ),
735+ )
619736 .handle("list", ({ path }) =>
620737 withWorkspaceRequestActor("sources.list", path.workspaceId, () =>
621738 withPolicy(requireReadSources(path.workspaceId))(
@@ -683,16 +800,6 @@ export const ControlPlaneSourcesLive = HttpApiBuilder.group( · 61 unmodified lines
683800 ),
684801 ),
685802 )
686− .handle("connectMcp", ({ path, payload }) =>
687− withWorkspaceRequestActor("sources.connectMcp", path.workspaceId, () =>
688− withPolicy(requireWriteSources(path.workspaceId))(
689− connectMcpSource({
690− workspaceId: path.workspaceId,
691− payload,
692− }),
693− ),
694− ),
695− )
696803 .handle("remove", ({ path }) =>
697804 withWorkspaceRequestActor("sources.remove", path.workspaceId, () =>
698805 withPolicy(requireWriteSources(path.workspaceId))(
packages/control-plane/src/runtime/sources-operations.ts+42−40
@@ -1,5 +1,4 @@
11 import type {
2− ConnectMcpSourcePayload,
32 CreateSourcePayload,
43 UpdateSourcePayload,
54 } from "../api/sources/api";
@@ -24,7 +23,6 @@ import { · 18 unmodified lines
2423 operationErrors,
2524 } from "./operation-errors";
2625 import { createDefaultSecretMaterialResolver } from "./secret-material-providers";
27−import { RuntimeSourceAuthServiceTag } from "./source-auth-service";
2826 import { ControlPlaneStore, type ControlPlaneStoreShape } from "./store";
2927 import { syncSourceToolArtifacts } from "./tool-artifacts";
3028 import {
@@ -39,10 +37,17 @@ const sourceOps = { · 8 unmodified lines
3937 create: operationErrors("sources.create"),
4038 get: operationErrors("sources.get"),
4139 update: operationErrors("sources.update"),
42− connectMcp: operationErrors("sources.connectMcp"),
4340 remove: operationErrors("sources.remove"),
4441 } as const;
4542
43+const shouldAutoProbeSource = (source: Source): boolean =>
44+ source.enabled
45+ && (
46+ (source.kind === "openapi" && !!source.specUrl)
47+ || source.kind === "graphql"
48+ )
49+ && (source.status === "draft" || source.status === "probing");
50+
4651 const syncArtifactsForSource = (input: {
4752 store: ControlPlaneStoreShape;
4853 source: Source;
@@ -55,19 +60,50 @@ const syncArtifactsForSource = (input: { · 6 unmodified lines
5560 rows: input.store,
5661 });
5762
63+ // For HTTP-backed source kinds that can validate themselves from a remote
64+ // document, automatically attempt to probe and connect. This mirrors the
65+ // addExecutorSource flow by overriding status to "connected" so the sync
66+ // guard passes.
67+ const autoProbe = shouldAutoProbeSource(input.source);
68+ const sourceForSync = autoProbe
69+ ? { ...input.source, status: "connected" as const }
70+ : input.source;
71+
5872 const synced = yield* Effect.either(
5973 syncSourceToolArtifacts({
6074 rows: input.store,
61− source: input.source,
75+ source: sourceForSync,
6276 resolveSecretMaterial,
6377 }),
6478 );
6579
6680 return yield* Either.match(synced, {
67− onRight: () => Effect.succeed(input.source),
81+ onRight: () =>
82+ Effect.gen(function* () {
83+ if (autoProbe) {
84+ const connectedSource = yield* updateSourceFromPayload({
85+ source: input.source,
86+ payload: { status: "connected", lastError: null },
87+ now: Date.now(),
88+ }).pipe(
89+ Effect.mapError((cause) =>
90+ input.operation.badRequest(
91+ "Failed updating source status",
92+ cause instanceof Error ? cause.message : String(cause),
93+ ),
94+ ),
95+ );
96+ yield* mapPersistenceError(
97+ input.operation.child("source_connected"),
98+ persistSource(input.store, connectedSource),
99+ );
100+ return connectedSource;
101+ }
102+ return input.source;
103+ }),
68104 onLeft: (error) =>
69105 Effect.gen(function* () {
70− if (input.source.enabled && input.source.status === "connected") {
106+ if (autoProbe || (input.source.enabled && input.source.status === "connected")) {
71107 const erroredSource = yield* updateSourceFromPayload({
72108 source: input.source,
73109 payload: {
@@ -213,40 +249,6 @@ export const updateSource = (input: { · 139 unmodified lines
213249 });
214250 }));
215251
216−export const connectMcpSource = (input: {
217− workspaceId: WorkspaceId;
218− payload: ConnectMcpSourcePayload;
219−}) =>
220− Effect.gen(function* () {
221− const sourceAuthService = yield* RuntimeSourceAuthServiceTag;
222−
223− return yield* sourceAuthService.connectMcpSource({
224− workspaceId: input.workspaceId,
225− sourceId: input.payload.sourceId,
226− endpoint: input.payload.endpoint,
227− name: input.payload.name,
228− namespace: input.payload.namespace,
229− enabled: input.payload.enabled,
230− transport: input.payload.transport,
231− queryParams: input.payload.queryParams,
232− headers: input.payload.headers,
233− }).pipe(
234− Effect.mapError((cause) => {
235− if (cause instanceof Error && cause.message.startsWith("Source not found:")) {
236− return sourceOps.connectMcp.notFound(
237− "Source not found",
238− `workspaceId=${input.workspaceId} sourceId=${input.payload.sourceId}`,
239− );
240− }
241−
242− return sourceOps.connectMcp.unknownStorage(
243− cause,
244− "Failed preparing MCP source OAuth flow",
245− );
246− }),
247− );
248− });
249−
250252 export const removeSource = (input: {
251253 workspaceId: WorkspaceId;
252254 sourceId: SourceId;