From 575d9e9ab29b5f9dd1ea56fbc590ccbf7a297282 Mon Sep 17 00:00:00 2001 From: George Pickett <297992784+georgeatparallel@users.noreply.github.com> Date: Sun, 11 Oct 2026 03:17:25 +0000 Subject: [PATCH 1/3] feat(connectors): add anonymous Parallel Search MCP search and fetch --- connectors/ids.lock.json | 2 + connectors/parallel/README.md | 32 ++ .../parallel/endpoints/fetch/endpoint.ts | 38 +++ .../parallel/endpoints/fetch/schema/inputs.ts | 20 ++ .../parallel/endpoints/search/endpoint.ts | 33 ++ .../endpoints/search/schema/inputs.ts | 13 + connectors/parallel/fixtures/fetch.json | 169 ++++++++++ connectors/parallel/fixtures/search.json | 175 +++++++++++ connectors/parallel/provider.test.ts | 295 ++++++++++++++++++ connectors/parallel/provider.ts | 200 ++++++++++++ connectors/parallel/schema/common.ts | 10 + .../add-connector-parallel/proposal.md | 28 ++ .../specs/parallel-connector/spec.md | 39 +++ .../changes/add-connector-parallel/tasks.md | 6 + shared/testing/fixtures.ts | 3 + 15 files changed, 1063 insertions(+) create mode 100644 connectors/parallel/README.md create mode 100644 connectors/parallel/endpoints/fetch/endpoint.ts create mode 100644 connectors/parallel/endpoints/fetch/schema/inputs.ts create mode 100644 connectors/parallel/endpoints/search/endpoint.ts create mode 100644 connectors/parallel/endpoints/search/schema/inputs.ts create mode 100644 connectors/parallel/fixtures/fetch.json create mode 100644 connectors/parallel/fixtures/search.json create mode 100644 connectors/parallel/provider.test.ts create mode 100644 connectors/parallel/provider.ts create mode 100644 connectors/parallel/schema/common.ts create mode 100644 openspec/changes/add-connector-parallel/proposal.md create mode 100644 openspec/changes/add-connector-parallel/specs/parallel-connector/spec.md create mode 100644 openspec/changes/add-connector-parallel/tasks.md diff --git a/connectors/ids.lock.json b/connectors/ids.lock.json index 211f7568..605c1a76 100644 --- a/connectors/ids.lock.json +++ b/connectors/ids.lock.json @@ -488,6 +488,8 @@ "opoint#search-by-ids", "opoint#search-headlines", "opoint#suggest", + "parallel#fetch", + "parallel#search", "pdl#v5/company/enrich", "pdl#v5/company/search", "pdl#v5/person/enrich", diff --git a/connectors/parallel/README.md b/connectors/parallel/README.md new file mode 100644 index 00000000..b6dfc805 --- /dev/null +++ b/connectors/parallel/README.md @@ -0,0 +1,32 @@ +# Parallel Search MCP + +Free web search and page extraction with no Parallel account or API key. Search +uses Fast mode on the anonymous +[Search MCP endpoint](https://docs.parallel.ai/integrations/mcp/search-mcp). The +free tier is rate limited and intended for exploration and light use. + +From the repository root, with Deno 2.x installed: + +```bash +deno task check && deno task test +deno task catalog inspect 'parallel#search' +deno task engine:run 'parallel#search' --body '{"objective":"Find the official Deno 2 release announcement","search_queries":["Deno 2 release official"],"session_id":"5b160d4d-4dce-4ed3-a459-2eafdd467a47"}' +deno task engine:run 'parallel#fetch' --body '{"urls":["https://deno.com/blog/v2.0"],"objective":"Deno 2 release features","session_id":"5b160d4d-4dce-4ed3-a459-2eafdd467a47"}' +``` + +Generate a new `session_id` once per conversation and reuse it for related +search and fetch calls. `model_name` is optional; supply the exact runtime model +identifier only when known. `parallel#fetch` accepts up to 20 HTTP/HTTPS URLs, +an optional objective (up to 200 characters), `search_queries`, `full_content` +and `allow_live_fetch`. Prefer focused excerpts; full markdown can be large. +Fetch returns successful results and per-URL errors. + +The connector returns each tool's `structuredContent` directly. Both endpoints +use the FREE usage model and do not forward vendor credentials. Each engine run +opens its own Streamable HTTP MCP session and terminates it best-effort. JSON +and finite SSE POST replies are supported. There is no persistent stream, SSE +reconnection or automatic retry. Request timeout is 30 seconds. + +Replay tests run offline. The two success fixtures are trimmed recordings of +anonymous calls on 2026-10-11, with transport session IDs replaced by a fixed +placeholder. Protocol failures and SSE responses use synthetic test responses. diff --git a/connectors/parallel/endpoints/fetch/endpoint.ts b/connectors/parallel/endpoints/fetch/endpoint.ts new file mode 100644 index 00000000..1455d4c2 --- /dev/null +++ b/connectors/parallel/endpoints/fetch/endpoint.ts @@ -0,0 +1,38 @@ +import { defineEndpoint } from "@shared/core"; +import { z } from "zod"; +import { zFetchBody } from "./schema/inputs.ts"; + +export default defineEndpoint({ + endpoint: "/fetch", + meta: { + displayName: "Parallel Web Fetch", + summary: + "Extract focused excerpts or full markdown from up to 20 URLs, free without a key.", + description: + "Fetch specific HTTP/HTTPS pages and return focused excerpts " + + "with source URLs, plus per-URL errors. Supply an objective and " + + "optional search queries to focus extraction. Request full_content " + + "only when you need complete markdown; responses can be large.", + categories: ["web-scraping"], + }, + request: { method: "POST", path: "/mcp" }, + input: { + schema: { + body: zFetchBody.extend({ + urls: z.array( + z.url({ protocol: /^https?$/ }).regex(/^https?:\/\//), + ).min(1).max(20), + objective: zFetchBody.shape.objective.unwrap().unwrap().max(200) + .nullable().optional(), + }), + }, + toRequest: ({ data }) => ({ + body: { + jsonrpc: "2.0", + id: 2, + method: "tools/call", + params: { name: "web_fetch", arguments: data.input.body }, + }, + }), + }, +}); diff --git a/connectors/parallel/endpoints/fetch/schema/inputs.ts b/connectors/parallel/endpoints/fetch/schema/inputs.ts new file mode 100644 index 00000000..96b0cccc --- /dev/null +++ b/connectors/parallel/endpoints/fetch/schema/inputs.ts @@ -0,0 +1,20 @@ +import { z } from "zod"; +import { conversationFields } from "../../../schema/common.ts"; + +// Mirror of web_fetch tools/list (2026-10-11); limits live at the binding. +export const zFetchBody = z.strictObject({ + urls: z.array(z.string()).describe("HTTP/HTTPS URLs to extract, up to 20."), + objective: z.string().nullable().describe( + "What to extract; up to 200 characters.", + ).optional(), + search_queries: z.array(z.string()).nullable().describe( + "Queries that surfaced these URLs.", + ).optional(), + full_content: z.boolean().describe( + "Request full markdown as well as focused excerpts; may be large.", + ).optional(), + allow_live_fetch: z.boolean().describe( + "Allow fresh live fetching (server default true). False uses cached content only.", + ).optional(), + ...conversationFields, +}); diff --git a/connectors/parallel/endpoints/search/endpoint.ts b/connectors/parallel/endpoints/search/endpoint.ts new file mode 100644 index 00000000..96508fe4 --- /dev/null +++ b/connectors/parallel/endpoints/search/endpoint.ts @@ -0,0 +1,33 @@ +import { defineEndpoint } from "@shared/core"; +import { zSearchBody } from "./schema/inputs.ts"; + +export default defineEndpoint({ + endpoint: "/search", + meta: { + displayName: "Parallel Web Search", + summary: + "Search the web for ranked sources and focused excerpts, free without a key.", + description: + "Search the live web in Fast mode with a natural-language " + + "objective and related keyword queries. Returns ranked URLs, titles " + + "and excerpts. Use parallel#fetch when excerpts are insufficient " + + "or you need a specific page's content.", + categories: ["web-search"], + }, + request: { method: "POST", path: "/mcp" }, + input: { + schema: { + body: zSearchBody.extend({ + search_queries: zSearchBody.shape.search_queries.min(1), + }), + }, + toRequest: ({ data }) => ({ + body: { + jsonrpc: "2.0", + id: 2, + method: "tools/call", + params: { name: "web_search", arguments: data.input.body }, + }, + }), + }, +}); diff --git a/connectors/parallel/endpoints/search/schema/inputs.ts b/connectors/parallel/endpoints/search/schema/inputs.ts new file mode 100644 index 00000000..6b37a432 --- /dev/null +++ b/connectors/parallel/endpoints/search/schema/inputs.ts @@ -0,0 +1,13 @@ +import { z } from "zod"; +import { conversationFields } from "../../../schema/common.ts"; + +// Mirror of web_search tools/list (2026-10-11), no authored defaults. +export const zSearchBody = z.strictObject({ + objective: z.string().describe( + "Focused natural-language search objective.", + ), + search_queries: z.array(z.string()).describe( + "Related keyword queries; at least one, preferably 2–3.", + ), + ...conversationFields, +}); diff --git a/connectors/parallel/fixtures/fetch.json b/connectors/parallel/fixtures/fetch.json new file mode 100644 index 00000000..531b06eb --- /dev/null +++ b/connectors/parallel/fixtures/fetch.json @@ -0,0 +1,169 @@ +{ + "name": "parallel-fetch-recorded", + "description": "Recorded anonymous MCP initialization, notification, tool result and session termination, 2026-10-11. Session header replaced with fixture-session; responses trimmed and scrubbed.", + "calls": [ + { + "req": { + "method": "POST", + "url": "https://search.parallel.ai/mcp", + "body": { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": "2025-03-26", + "capabilities": {}, + "clientInfo": { + "name": "monid", + "version": "1.0.0" + } + } + } + }, + "res": { + "status": 200, + "headers": { + "mcp-session-id": "fixture-session" + }, + "body": { + "jsonrpc": "2.0", + "id": 1, + "result": { + "protocolVersion": "2025-03-26", + "capabilities": { + "experimental": {}, + "prompts": { + "listChanged": false + }, + "resources": { + "subscribe": false, + "listChanged": false + }, + "tools": { + "listChanged": false + } + }, + "serverInfo": { + "name": "Parallel Web Search MCP Server", + "version": "1.27.0", + "websiteUrl": "https://parallel.ai", + "icons": [ + { + "src": "https://assets.parallel.ai/dark-parallel-avatar-270.svg", + "mimeType": "image/svg+xml", + "theme": "light" + }, + { + "src": "https://assets.parallel.ai/dark-parallel-avatar-270.png", + "mimeType": "image/png", + "sizes": [ + "270x270" + ], + "theme": "light" + } + ] + }, + "instructions": "\nThe search and fetch tools are read-only, idempotent, and safe to call\nconcurrently. When several independent search or fetch objectives are already\nknown, issue one tool call per objective concurrently; do not wait for one\nresult before starting another. Batch related search queries or URLs that share\na common objective into one call.\n\n\n\nUse web_search first for most factual, current-information, research,\ncomparison, documentation, and troubleshooting questions.\n\nThe web_search results includ" + } + } + } + }, + { + "req": { + "method": "POST", + "url": "https://search.parallel.ai/mcp", + "body": { + "jsonrpc": "2.0", + "method": "notifications/initialized" + } + }, + "res": { + "status": 202, + "headers": { + "mcp-session-id": "fixture-session" + }, + "body": null + } + }, + { + "req": { + "method": "POST", + "url": "https://search.parallel.ai/mcp", + "body": { + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": { + "name": "web_fetch", + "arguments": { + "urls": [ + "https://deno.com/blog/v2.0" + ], + "objective": "Deno 2 release features", + "session_id": "c60dac7f-d33a-4643-bd06-7f7bf514b1f8" + } + } + } + }, + "res": { + "status": 200, + "headers": { + "mcp-session-id": "fixture-session" + }, + "body": { + "jsonrpc": "2.0", + "id": 2, + "result": { + "_meta": { + "parallel/usage": [ + { + "name": "sku_extract_excerpts", + "count": 1, + "cost_usd": 0.001 + } + ] + }, + "content": [ + { + "type": "text", + "text": "{\n \"extract_id\": \"extract_35ba304a689e504f491225a3be81a805\",\n \"results\": [\n {\n \"url\": \"https://deno.com/blog/v2.0\",\n \"title\": \"Announcing Deno 2 | Deno\",\n \"publish_date\": null,\n \"excerpts\": [\n \"Introducing celld: self-hosted, distributed Durable Objects\\n[Learn more ->](https://celld.dev/)\\nDismiss\\n[](https://deno.com/)\\n`⌘K`\\n`↑↓ Up or down to` navigate `↵ Enter to` select `ESC Escape to` close Search powered by Orama Search powered by Orama\\n`⌘K`\\n\\n...\\n\\n# An" + } + ], + "structuredContent": { + "extract_id": "extract_35ba304a689e504f491225a3be81a805", + "results": [ + { + "url": "https://deno.com/blog/v2.0", + "title": "Announcing Deno 2 | Deno", + "publish_date": null, + "excerpts": [ + "Introducing celld: self-hosted, distributed Durable Objects\n[Learn more ->](https://celld.dev/)\nDismiss\n[](https://deno.com/)\n`⌘K`\n`↑↓ Up or down to` navigate `↵ Enter to` select `ESC Escape to` close Search powered by Orama Search powered by Orama\n`⌘K`\n\n...\n\n# Announcing Deno 2\n## Announcing Deno 2\n* **Native TypeScript support**\n* **Built on web standards** : Promises, fetch, and ES Modules\n* **Batteries included** : builtin formatter, linter, type checker, test\n framework, compile to executa" + ], + "full_content": null + } + ], + "errors": [], + "warnings": null, + "metadata": null, + "session_id": "c60dac7f-d33a-4643-bd06-7f7bf514b1f8" + }, + "isError": false + } + } + } + }, + { + "req": { + "method": "DELETE", + "url": "https://search.parallel.ai/mcp" + }, + "res": { + "status": 200, + "headers": { + "mcp-session-id": "fixture-session" + }, + "body": null + } + } + ] +} diff --git a/connectors/parallel/fixtures/search.json b/connectors/parallel/fixtures/search.json new file mode 100644 index 00000000..166d3c41 --- /dev/null +++ b/connectors/parallel/fixtures/search.json @@ -0,0 +1,175 @@ +{ + "name": "parallel-search-recorded", + "description": "Recorded anonymous MCP initialization, notification, tool result and session termination, 2026-10-11. Session header replaced with fixture-session; responses trimmed and scrubbed.", + "calls": [ + { + "req": { + "method": "POST", + "url": "https://search.parallel.ai/mcp", + "body": { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": "2025-03-26", + "capabilities": {}, + "clientInfo": { + "name": "monid", + "version": "1.0.0" + } + } + } + }, + "res": { + "status": 200, + "headers": { + "mcp-session-id": "fixture-session" + }, + "body": { + "jsonrpc": "2.0", + "id": 1, + "result": { + "protocolVersion": "2025-03-26", + "capabilities": { + "experimental": {}, + "prompts": { + "listChanged": false + }, + "resources": { + "subscribe": false, + "listChanged": false + }, + "tools": { + "listChanged": false + } + }, + "serverInfo": { + "name": "Parallel Web Search MCP Server", + "version": "1.27.0", + "websiteUrl": "https://parallel.ai", + "icons": [ + { + "src": "https://assets.parallel.ai/dark-parallel-avatar-270.svg", + "mimeType": "image/svg+xml", + "theme": "light" + }, + { + "src": "https://assets.parallel.ai/dark-parallel-avatar-270.png", + "mimeType": "image/png", + "sizes": [ + "270x270" + ], + "theme": "light" + } + ] + }, + "instructions": "\nThe search and fetch tools are read-only, idempotent, and safe to call\nconcurrently. When several independent search or fetch objectives are already\nknown, issue one tool call per objective concurrently; do not wait for one\nresult before starting another. Batch related search queries or URLs that share\na common objective into one call.\n\n\n\nUse web_search first for most factual, current-information, research,\ncomparison, documentation, and troubleshooting questions.\n\nThe web_search results includ" + } + } + } + }, + { + "req": { + "method": "POST", + "url": "https://search.parallel.ai/mcp", + "body": { + "jsonrpc": "2.0", + "method": "notifications/initialized" + } + }, + "res": { + "status": 202, + "headers": { + "mcp-session-id": "fixture-session" + }, + "body": null + } + }, + { + "req": { + "method": "POST", + "url": "https://search.parallel.ai/mcp", + "body": { + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": { + "name": "web_search", + "arguments": { + "objective": "Find the official Deno 2 release announcement", + "search_queries": [ + "Deno 2 release official" + ], + "session_id": "c60dac7f-d33a-4643-bd06-7f7bf514b1f8" + } + } + } + }, + "res": { + "status": 200, + "headers": { + "mcp-session-id": "fixture-session" + }, + "body": { + "jsonrpc": "2.0", + "id": 2, + "result": { + "_meta": { + "parallel/usage": [ + { + "name": "sku_search", + "count": 1, + "cost_usd": 0.001 + } + ] + }, + "content": [ + { + "type": "text", + "text": "{\n \"search_id\": \"search_749ce92b31312340c980ea9e5171f7fe\",\n \"results\": [\n {\n \"url\": \"https://deno.com/blog/v2.0\",\n \"title\": \"Announcing Deno 2 | Deno\",\n \"publish_date\": null,\n \"excerpts\": [\n \"Announcing Deno 2\\nOctober 9, 2024\\n* Ryan Dahl\\n* Bert Belder\\n* Bartek Iwańczuk\\n* Andy Jiang\\n* Product Update\\n Watch our video announcement.\\nAnnouncing Deno 2\\nIn an effort to simplify web programming, we created Deno: a modern, all-in-one, zero-config toolchain for Ja" + } + ], + "structuredContent": { + "search_id": "search_749ce92b31312340c980ea9e5171f7fe", + "results": [ + { + "url": "https://deno.com/blog/v2.0", + "title": "Announcing Deno 2 | Deno", + "publish_date": null, + "excerpts": [ + "Announcing Deno 2\nOctober 9, 2024\n* Ryan Dahl\n* Bert Belder\n* Bartek Iwańczuk\n* Andy Jiang\n* Product Update\n Watch our video announcement.\nAnnouncing Deno 2\nIn an effort to simplify web programming, we created Deno: a modern, all-in-one, zero-config toolchain for JavaScript and TypeScript development.\nThis means seamless interoperability with legacy JavaScript infrastructure and support for a wider range of projects and development teams. All without sacrificing the simplicity, security, and “b" + ] + }, + { + "url": "https://deno.com/blog?tag=product-update", + "title": "Blog | Deno", + "publish_date": null, + "excerpts": [ + "Deno 2.9 deno desktop for building native desktop apps from web tech, first-class migration from npm/pnpm/yarn/Bun, CSS module imports, snapshot and parameterized testing, smaller deno compile --bundle binaries, and Node.js 26 compatibility. > Deno 2.8 import defer, six new subcommands (deno transpile, deno pack, deno bump-version, deno ci, deno why, deno audit fix), network debugging in Chrome DevTools, framework-aware deno compile, and 3.66x faster cold npm installs. > Deno 2.0 Release Candida" + ] + } + ], + "warnings": null, + "metadata": null, + "session_id": "c60dac7f-d33a-4643-bd06-7f7bf514b1f8" + }, + "isError": false + } + } + } + }, + { + "req": { + "method": "DELETE", + "url": "https://search.parallel.ai/mcp" + }, + "res": { + "status": 200, + "headers": { + "mcp-session-id": "fixture-session" + }, + "body": null + } + } + ] +} diff --git a/connectors/parallel/provider.test.ts b/connectors/parallel/provider.test.ts new file mode 100644 index 00000000..ad181701 --- /dev/null +++ b/connectors/parallel/provider.test.ts @@ -0,0 +1,295 @@ +import { assert, assertEquals, assertRejects } from "@std/assert"; +import { directTransport, Engine } from "@monid/connector-engine"; +import { + estimateEndpoint, + loadFixture, + runEndpoint, + testBundle, + testSealedUnit, +} from "@shared/testing"; +import type { Json } from "@shared/core"; + +const anonymousHeaders = (init?: RequestInit) => { + const headers = new Headers(init?.headers); + assertEquals( + headers.get("user-agent"), + "monid/parallel-search-mcp (https://github.com/monid-ai/monid)", + ); + assertEquals(headers.get("accept"), "application/json, text/event-stream"); + assertEquals(headers.get("authorization"), null); + assertEquals(headers.get("x-api-key"), null); + return headers; +}; + +async function execute( + options: { + stage?: string; + status?: number; + reply?: Json; + sse?: boolean; + cleanupThrows?: boolean; + session?: boolean; + } = {}, +) { + const methods: string[] = []; + const endpoint = await new Engine({ + transport: directTransport({ + params: (_provider, fields) => { + assertEquals(fields, ["apiKey"]); + return Promise.resolve({ apiKey: "saved-key-must-not-egress" }); + }, + fetch: (url, init) => { + assertEquals(String(url), "https://search.parallel.ai/mcp"); + const headers = anonymousHeaders(init); + const request = init?.body ? JSON.parse(String(init.body)) : {}; + const method = init?.method === "DELETE" + ? "DELETE" + : request.method; + methods.push(method); + if (method !== "initialize") { + assertEquals( + headers.get("mcp-protocol-version"), + "2025-03-26", + ); + assertEquals( + headers.get("mcp-session-id"), + options.session === false ? null : "test-session", + ); + } + if (method === "DELETE") { + if (options.cleanupThrows) { + return Promise.reject(new Error("cleanup failed")); + } + return Promise.resolve(new Response(null, { status: 405 })); + } + if (method === options.stage) { + return Promise.resolve( + new Response( + JSON.stringify( + options.reply ?? { message: "limited" }, + ), + { status: options.status ?? 200 }, + ), + ); + } + if (method === "notifications/initialized") { + assertEquals(request.id, undefined); + return Promise.resolve(new Response(null, { status: 202 })); + } + const result = method === "initialize" + ? { + protocolVersion: "2025-03-26", + capabilities: { tools: {} }, + serverInfo: { name: "fixture", version: "1" }, + } + : { + structuredContent: { + results: [{ + url: "https://deno.com/blog/v2.0", + excerpts: [ + "Deno 2 is backward compatible with Node.js.", + ], + }], + }, + }; + if (method === "tools/call") { + assertEquals(request.params.name, "web_search"); + assertEquals(request.params.arguments, { + objective: "Deno 2", + search_queries: ["Deno 2 release"], + }); + } + const reply = { jsonrpc: "2.0", id: request.id, result }; + // A comment, a progress notification and a multiline response. + const wire = options.sse + ? ': keepalive\r\n\r\ndata: {"jsonrpc":"2.0","method":"notifications/progress"}\r\n\r\nevent: message\r\ndata: ' + + JSON.stringify(reply).slice(0, -1) + + "\r\ndata: }\r\n\r\n" + : JSON.stringify(reply); + return Promise.resolve( + new Response(wire, { + status: 200, + headers: + method === "initialize" && options.session !== false + ? { "mcp-session-id": "test-session" } + : {}, + }), + ); + }, + }), + }).load(await testSealedUnit("parallel#search")); + const result = await endpoint.run({ + body: { objective: "Deno 2", search_queries: ["Deno 2 release"] }, + }); + return { result, methods }; +} + +Deno.test("parallel: catalog identities, empty credentials and FREE estimates", async () => { + const bundle = await testBundle(); + for (const tool of ["search", "fetch"]) { + const unit = await testSealedUnit(`parallel#${tool}`); + assertEquals(unit.doc.request.url, "https://search.parallel.ai/mcp"); + assertEquals(unit.doc.usage.model.kind, "FREE"); + assertEquals(unit.doc.auth.credentials.properties, {}); + assertEquals( + await estimateEndpoint(unit, { + body: tool === "search" + ? { objective: "Deno", search_queries: ["Deno"] } + : { urls: ["https://deno.com"] }, + }), + { credits: {}, evidence: {} }, + ); + const lock = JSON.parse( + await Deno.readTextFile( + new URL("../ids.lock.json", import.meta.url), + ), + ); + assert(lock.endpoints.includes(`parallel#${tool}`)); + } + assertEquals( + bundle.endpoints["parallel#search"].lifecycle?.start, + bundle.endpoints["parallel#fetch"].lifecycle?.start, + ); + // Existing reference provider still selects its native keyed route. + assertEquals( + bundle.endpoints["exa#search"].request.url, + "https://api.exa.ai/search", + ); +}); + +for (const sse of [false, true]) { + Deno.test(`parallel: compiled caller ${sse ? "SSE" : "JSON"} replies and best-effort cleanup`, async () => { + const { result, methods } = await execute({ sse, cleanupThrows: true }); + assertEquals(methods, [ + "initialize", + "notifications/initialized", + "tools/call", + "DELETE", + ]); + assertEquals(result.isProviderError, false); + assertEquals(result.usage, { credits: {}, evidence: {} }); + assert( + Array.isArray((result.output as { results: unknown[] }).results), + ); + }); +} + +Deno.test("parallel: stateless server omits transport session", async () => { + const { result, methods } = await execute({ session: false }); + assertEquals(methods, [ + "initialize", + "notifications/initialized", + "tools/call", + ]); + assertEquals(result.isProviderError, false); +}); + +for (const stage of ["initialize", "notifications/initialized", "tools/call"]) { + Deno.test(`parallel: ${stage} 429 is provider error, no retry, zero usage`, async () => { + const { result, methods } = await execute({ stage, status: 429 }); + assertEquals(result.httpStatus, 429); + assertEquals(result.isProviderError, true); + assertEquals(result.usage, { credits: {}, evidence: {} }); + assertEquals(methods.filter((m) => m === stage).length, 1); + if (stage !== "tools/call") { + assertEquals(methods.includes("tools/call"), false); + } + }); +} + +for ( + const reply of [ + { + jsonrpc: "2.0", + id: 2, + error: { code: -32602, message: "Invalid input" }, + }, + { + jsonrpc: "2.0", + id: 2, + result: { + isError: true, + content: [{ type: "text", text: "Failed" }], + }, + }, + { jsonrpc: "2.0", id: 99, result: {} }, + { jsonrpc: "2.0", id: 2, result: {} }, + "not valid SSE", + ] as Json[] +) { + Deno.test(`parallel: invalid tool reply ${JSON.stringify(reply)} fails as data`, async () => { + const { result, methods } = await execute({ + stage: "tools/call", + reply, + }); + assertEquals(result.httpStatus, 502); + assertEquals(result.isProviderError, true); + assertEquals(result.usage, { credits: {}, evidence: {} }); + assertEquals(methods.at(-1), "DELETE"); + }); +} + +Deno.test("parallel: unsupported negotiated version prevents tool dispatch", async () => { + const { result, methods } = await execute({ + stage: "initialize", + reply: { + jsonrpc: "2.0", + id: 1, + result: { protocolVersion: "2099-01-01" }, + }, + }); + assertEquals(result.httpStatus, 502); + assertEquals(methods, ["initialize"]); +}); + +for ( + const [tool, bodies] of [ + ["search", [{ objective: "test", search_queries: [] }, { + search_queries: ["test"], + }, { objective: "test", search_queries: ["test"], api_key: "secret" }]], + ["fetch", [{ urls: [] }, { urls: ["file:///etc/passwd"] }, { + urls: Array(21).fill("https://deno.com"), + }, { urls: ["https://deno.com"], objective: "x".repeat(201) }]], + ] as [string, Json[]][] +) { + Deno.test(`parallel#${tool}: input limits reject before egress`, async () => { + const unit = await testSealedUnit(`parallel#${tool}`); + for (const body of bodies) { + await assertRejects( + () => runEndpoint({ unit, input: { body }, mode: "replay" }), + Error, + "INVALID_INPUT", + ); + } + }); +} + +for (const tool of ["search", "fetch"]) { + Deno.test(`parallel#${tool}: recorded anonymous protocol chain returns useful output`, async () => { + const fixture = await loadFixture( + new URL(`./fixtures/${tool}.json`, import.meta.url).pathname, + ); + const body = + (fixture.calls[2].req.body as { params: { arguments: Json } }) + .params.arguments; + const result = await runEndpoint({ + unit: await testSealedUnit(`parallel#${tool}`), + input: { body }, + mode: "replay", + fixture, + }); + assertEquals(result.isProviderError, false); + assertEquals(result.usage, { credits: {}, evidence: {} }); + const output = result.output as { + results: { url: string; excerpts: string[] }[]; + }; + assert(output.results.length > 0); + assert(output.results[0].excerpts[0].length > 30); + assertEquals( + result.output, + (fixture.calls[2].res.body as { + result: { structuredContent: Json }; + }).result.structuredContent, + ); + }); +} diff --git a/connectors/parallel/provider.ts b/connectors/parallel/provider.ts new file mode 100644 index 00000000..3e7c9941 --- /dev/null +++ b/connectors/parallel/provider.ts @@ -0,0 +1,200 @@ +import { + defineProvider, + type LifecycleStartFn, + UsageModelKind, +} from "@shared/core"; +import { z } from "zod"; + +// All IO uses the existing transport, including its request timeout. The +// endpoint's toRequest builds the tool envelope; this hook owns the session. +const start: LifecycleStartFn = async ({ data, utils }) => { + const headers: Record = { + "Accept": "application/json, text/event-stream", + "User-Agent": + "monid/parallel-search-mcp (https://github.com/monid-ai/monid)", + }; + let session; + try { + for (const stage of ["initialize", "tools/call"]) { + const id = stage === "initialize" ? 1 : 2; + const res = await utils.http({ + method: "POST", + url: data.request.url, + headers, + body: stage === "initialize" + ? { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { + protocolVersion: "2025-03-26", + capabilities: {}, + clientInfo: { name: "monid", version: "1.0.0" }, + }, + } + : data.input.body, + }); + if (res.status < 200 || res.status >= 300) { + return { + kind: "COMPLETED", + httpStatus: res.status, + output: res.body, + }; + } + // The transport buffers the response. Parallel normally returns + // JSON; finite SSE POST replies may contain progress notifications + // before the matching response. Join multiline data per SSE rules. + let reply; + try { + const messages = typeof res.body === "string" + ? res.body.replace(/\r\n/g, "\n").split("\n\n") + .map((event) => + event.split("\n") + .filter((line) => line.startsWith("data:")) + .map((line) => line.slice(5).replace(/^ /, "")) + .join("\n") + ) + .filter((payload) => payload !== "") + .map((payload) => JSON.parse(payload)) + : [res.body]; + reply = messages.find((message) => + utils.json.optionalGet(message, "$.jsonrpc") === "2.0" && + utils.json.optionalGet(message, "$.id") === id + ); + } catch { + return { + kind: "COMPLETED", + httpStatus: 502, + output: { message: "Malformed MCP response" }, + }; + } + if (reply === undefined) { + return { + kind: "COMPLETED", + httpStatus: 502, + output: { message: "Missing matching MCP reply" }, + }; + } + const error = utils.json.optionalGet(reply, "$.error"); + if (error !== undefined) { + return { kind: "COMPLETED", httpStatus: 502, output: error }; + } + const result = utils.json.optionalGet(reply, "$.result"); + if ( + result === null || typeof result !== "object" || + Array.isArray(result) + ) { + return { + kind: "COMPLETED", + httpStatus: 502, + output: { message: "Missing MCP result" }, + }; + } + if (stage === "initialize") { + if ( + utils.json.optionalGet(result, "$.protocolVersion") !== + "2025-03-26" + ) { + return { + kind: "COMPLETED", + httpStatus: 502, + output: { message: "Unsupported MCP protocol version" }, + }; + } + session = res.headers["mcp-session-id"]; + if (session) headers["Mcp-Session-Id"] = session; + headers["MCP-Protocol-Version"] = "2025-03-26"; + const ready = await utils.http({ + method: "POST", + url: data.request.url, + headers, + body: { + jsonrpc: "2.0", + method: "notifications/initialized", + }, + }); + if (ready.status < 200 || ready.status >= 300) { + return { + kind: "COMPLETED", + httpStatus: ready.status, + output: ready.body, + }; + } + } else { + if (utils.json.optionalGet(result, "$.isError") === true) { + return { + kind: "COMPLETED", + httpStatus: 502, + output: result, + }; + } + const structured = utils.json.optionalGet( + result, + "$.structuredContent", + ); + if ( + structured === null || typeof structured !== "object" || + Array.isArray(structured) + ) { + return { + kind: "COMPLETED", + httpStatus: 502, + output: { + message: "Missing MCP structured content", + raw: result, + }, + }; + } + return { + kind: "COMPLETED", + httpStatus: 200, + output: structured, + }; + } + } + throw new Error("MCP tool call did not complete"); + } finally { + if (session) { + // MCP session termination is optional server-side. Cleanup errors + // must not replace a useful result or the original failure. + try { + await utils.http({ + method: "DELETE", + url: data.request.url, + headers, + }); + } catch { /* best-effort */ } + } + } +}; + +export default defineProvider({ + name: "parallel", + meta: { + displayName: "Parallel", + summary: + "Free web search and page extraction through Parallel Search MCP.", + description: + "Anonymous web search in Fast mode and focused page excerpts " + + "through Parallel Search MCP. No Parallel account or API key required.", + homepageUrl: "https://parallel.ai", + docsUrl: "https://docs.parallel.ai/integrations/mcp/search-mcp", + categories: ["web-search"], + notes: [ + "The anonymous free tier is rate limited and intended for exploration and light use.", + "Reuse session_id across related search and fetch calls in a conversation. This is separate from the per-run MCP transport session.", + ], + }, + // The contract requires an injector. An empty shape requests no secrets; + // the identity injector keeps every owned request anonymous. The loose + // shape tolerates the local resolver's legacy apiKey fallback, ignoring + // any saved key instead of requiring or forwarding it. + auth: { + credentials: z.looseObject({}), + inject: ({ data }) => data.request, + }, + request: { baseUrl: "https://search.parallel.ai" }, + timeouts: { requestMs: 30_000, runMs: 120_000 }, + usage: { model: { kind: UsageModelKind.FREE } }, + lifecycle: { start }, +}); diff --git a/connectors/parallel/schema/common.ts b/connectors/parallel/schema/common.ts new file mode 100644 index 00000000..6abb4daa --- /dev/null +++ b/connectors/parallel/schema/common.ts @@ -0,0 +1,10 @@ +import { z } from "zod"; + +export const conversationFields = { + session_id: z.string().max(100).describe( + "Stable conversation identifier. Generate once and reuse for related search/fetch calls.", + ).optional(), + model_name: z.string().max(100).describe( + "Exact model identifier from trusted runtime configuration; omit if unknown.", + ).optional(), +}; diff --git a/openspec/changes/add-connector-parallel/proposal.md b/openspec/changes/add-connector-parallel/proposal.md new file mode 100644 index 00000000..76f15ec5 --- /dev/null +++ b/openspec/changes/add-connector-parallel/proposal.md @@ -0,0 +1,28 @@ +# Proposal: add-connector-parallel + +## Why + +Add anonymous web search and page extraction through Parallel Search MCP, +using the existing lifecycle HTTP port. No vendor API key is needed. + +## What Changes + +- Add `parallel#search` and `parallel#fetch`, both FREE, with faithful MCP + tool inputs and no credential fields. +- Initialize a Streamable HTTP session per run, notify initialized, invoke + the selected tool, and terminate the session best-effort. +- Return structured tool results, preserving MCP errors as provider errors. +- Add compiled-artifact tests, observed request assertions and recorded fixtures. + +## Capabilities + +- `parallel-connector`. + +## Non-goals + +Paid REST APIs, authentication, connection-level overrides, persistent MCP +connections, server-initiated requests, SSE reconnection and agent orchestration. + +## Impact + +Connector-only change. Existing defaults, engine, schema and ABI are unchanged. diff --git a/openspec/changes/add-connector-parallel/specs/parallel-connector/spec.md b/openspec/changes/add-connector-parallel/specs/parallel-connector/spec.md new file mode 100644 index 00000000..f92cf9a5 --- /dev/null +++ b/openspec/changes/add-connector-parallel/specs/parallel-connector/spec.md @@ -0,0 +1,39 @@ +# Parallel connector + +## ADDED Requirements + +### Requirement: Anonymous search and fetch + +The catalog SHALL expose `parallel#search` and `parallel#fetch` through +`https://search.parallel.ai/mcp`, without credential fields or auth headers. +Both endpoints SHALL use the FREE usage model. + +#### Scenario: User selects search + +- **WHEN** the caller supplies an objective and related search queries +- **THEN** the connector invokes `web_search` and returns its structured results. + +#### Scenario: User selects fetch + +- **WHEN** the caller supplies up to 20 HTTP/HTTPS URLs +- **THEN** the connector invokes `web_fetch` and returns results and per-URL errors. + +### Requirement: MCP protocol and errors + +Each run SHALL initialize the supported protocol, propagate the session header +when provided, send the initialized notification, and invoke the selected tool. +All requests SHALL identify Monid with a project User-Agent. The connector SHALL +accept JSON and finite SSE POST responses and match JSON-RPC reply IDs. +HTTP failures, JSON-RPC errors, tool errors and malformed replies SHALL become +provider errors with zero usage. Session termination SHALL be best-effort and +SHALL NOT replace the tool outcome. Per-request timeouts SHALL remain bounded. + +#### Scenario: Tool failure + +- **WHEN** a 200 reply contains `isError: true` or a JSON-RPC error +- **THEN** the run reports a provider error rather than successful search output. + +#### Scenario: Rate limited + +- **WHEN** a protocol request returns 429 +- **THEN** the run returns the provider failure without automatic retries. diff --git a/openspec/changes/add-connector-parallel/tasks.md b/openspec/changes/add-connector-parallel/tasks.md new file mode 100644 index 00000000..4fa75359 --- /dev/null +++ b/openspec/changes/add-connector-parallel/tasks.md @@ -0,0 +1,6 @@ +# Tasks + +- [x] Add provider, tool schemas and endpoints. +- [x] Verify compiled artifact loading, protocol, errors and anonymous requests. +- [x] Record and replay successful search and fetch. +- [ ] Run formatting, typecheck and repository replay tests. diff --git a/shared/testing/fixtures.ts b/shared/testing/fixtures.ts index 8e369d36..a6271a1c 100644 --- a/shared/testing/fixtures.ts +++ b/shared/testing/fixtures.ts @@ -34,6 +34,9 @@ import { type Json, zJson } from "@shared/core"; * written. */ export const RECORDED_RES_HEADERS = [ + // Parallel MCP transport session. Recorded fixtures replace the live ID + // with a fixed placeholder; no reusable session identifier is committed. + "mcp-session-id", "location", "retry-after", "x-api-units-cost-total-actual", From f7cda463f0df736b0850b5414a305902b7858350 Mon Sep 17 00:00:00 2001 From: George Pickett <297992784+georgeatparallel@users.noreply.github.com> Date: Sun, 11 Oct 2026 03:19:02 +0000 Subject: [PATCH 2/3] fix(connectors): validate MCP envelopes and scrub recorded sessions --- .../parallel/endpoints/fetch/endpoint.ts | 2 +- .../parallel/endpoints/search/endpoint.ts | 5 +++- connectors/parallel/provider.test.ts | 9 +++++-- connectors/parallel/provider.ts | 8 +++--- shared/testing/fixture-size.test.ts | 27 +++++++++++++++++++ shared/testing/fixtures.ts | 7 ++++- 6 files changed, 50 insertions(+), 8 deletions(-) diff --git a/connectors/parallel/endpoints/fetch/endpoint.ts b/connectors/parallel/endpoints/fetch/endpoint.ts index 1455d4c2..26fb6d7d 100644 --- a/connectors/parallel/endpoints/fetch/endpoint.ts +++ b/connectors/parallel/endpoints/fetch/endpoint.ts @@ -31,7 +31,7 @@ export default defineEndpoint({ jsonrpc: "2.0", id: 2, method: "tools/call", - params: { name: "web_fetch", arguments: data.input.body }, + params: { name: "web_fetch", arguments: data.input.body ?? {} }, }, }), }, diff --git a/connectors/parallel/endpoints/search/endpoint.ts b/connectors/parallel/endpoints/search/endpoint.ts index 96508fe4..f31de1f8 100644 --- a/connectors/parallel/endpoints/search/endpoint.ts +++ b/connectors/parallel/endpoints/search/endpoint.ts @@ -26,7 +26,10 @@ export default defineEndpoint({ jsonrpc: "2.0", id: 2, method: "tools/call", - params: { name: "web_search", arguments: data.input.body }, + params: { + name: "web_search", + arguments: data.input.body ?? {}, + }, }, }), }, diff --git a/connectors/parallel/provider.test.ts b/connectors/parallel/provider.test.ts index ad181701..fb6c3d10 100644 --- a/connectors/parallel/provider.test.ts +++ b/connectors/parallel/provider.test.ts @@ -68,7 +68,12 @@ async function execute( JSON.stringify( options.reply ?? { message: "limited" }, ), - { status: options.status ?? 200 }, + { + status: options.status ?? 200, + headers: method === "initialize" + ? { "mcp-session-id": "test-session" } + : {}, + }, ), ); } @@ -239,7 +244,7 @@ Deno.test("parallel: unsupported negotiated version prevents tool dispatch", asy }, }); assertEquals(result.httpStatus, 502); - assertEquals(methods, ["initialize"]); + assertEquals(methods, ["initialize", "DELETE"]); }); for ( diff --git a/connectors/parallel/provider.ts b/connectors/parallel/provider.ts index 3e7c9941..03431d09 100644 --- a/connectors/parallel/provider.ts +++ b/connectors/parallel/provider.ts @@ -41,6 +41,11 @@ const start: LifecycleStartFn = async ({ data, utils }) => { output: res.body, }; } + if (stage === "initialize") { + session = res.headers["mcp-session-id"]; + if (session) headers["Mcp-Session-Id"] = session; + headers["MCP-Protocol-Version"] = "2025-03-26"; + } // The transport buffers the response. Parallel normally returns // JSON; finite SSE POST replies may contain progress notifications // before the matching response. Join multiline data per SSE rules. @@ -101,9 +106,6 @@ const start: LifecycleStartFn = async ({ data, utils }) => { output: { message: "Unsupported MCP protocol version" }, }; } - session = res.headers["mcp-session-id"]; - if (session) headers["Mcp-Session-Id"] = session; - headers["MCP-Protocol-Version"] = "2025-03-26"; const ready = await utils.http({ method: "POST", url: data.request.url, diff --git a/shared/testing/fixture-size.test.ts b/shared/testing/fixture-size.test.ts index 2eda132e..bc514cf4 100644 --- a/shared/testing/fixture-size.test.ts +++ b/shared/testing/fixture-size.test.ts @@ -265,3 +265,30 @@ Deno.test("zRecordedCall: response headers outside the allowlist are rejected", }; assertEquals(zRecordedCall.safeParse(smuggled).success, false); }); + +Deno.test("MCP recording: live session is forwarded for execution and redacted before disk", async () => { + const sink: Parameters[0] = []; + const recorded = recordingFetch(() => + Promise.resolve( + new Response("{}", { + headers: { + "mcp-session-id": "live-opaque-session-token", + "set-cookie": "secret", + }, + }), + ), sink); + const response = await recorded("https://search.parallel.ai/mcp", { + method: "POST", + }); + assertEquals( + response.headers.get("mcp-session-id"), + "live-opaque-session-token", + ); + assertEquals(response.headers.get("set-cookie"), null); + assertEquals(scrubCalls(sink)[0].res.headers, { + "mcp-session-id": "fixture-session", + }); + assertEquals(zRecordedCall.parse(scrubCalls(sink)[0]).res.headers, { + "mcp-session-id": "fixture-session", + }); +}); diff --git a/shared/testing/fixtures.ts b/shared/testing/fixtures.ts index a6271a1c..d9c7817f 100644 --- a/shared/testing/fixtures.ts +++ b/shared/testing/fixtures.ts @@ -236,7 +236,12 @@ export function scrubCalls(calls: RecordedCall[]): RecordedCall[] { headers: Object.fromEntries( Object.entries(call.res.headers).map(( [key, value], - ) => [key, scrubUrlCredentials(value)]), + ) => [ + key, + key.toLowerCase() === "mcp-session-id" + ? "fixture-session" + : scrubUrlCredentials(value), + ]), ), } : {}), From 9c52a383a9c4de6ca7a78d625dc50b81e599f815 Mon Sep 17 00:00:00 2001 From: George Pickett <297992784+georgeatparallel@users.noreply.github.com> Date: Sun, 11 Oct 2026 03:20:31 +0000 Subject: [PATCH 3/3] docs(openspec): record completed Parallel connector validation --- openspec/changes/add-connector-parallel/tasks.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/openspec/changes/add-connector-parallel/tasks.md b/openspec/changes/add-connector-parallel/tasks.md index 4fa75359..fe712376 100644 --- a/openspec/changes/add-connector-parallel/tasks.md +++ b/openspec/changes/add-connector-parallel/tasks.md @@ -3,4 +3,4 @@ - [x] Add provider, tool schemas and endpoints. - [x] Verify compiled artifact loading, protocol, errors and anonymous requests. - [x] Record and replay successful search and fetch. -- [ ] Run formatting, typecheck and repository replay tests. +- [x] Run formatting, typecheck and repository replay tests.