mirror of
https://github.com/anomalyco/opencode.git
synced 2026-11-07 16:17:25 +03:00
feat(server): let embedders supply the fetch for OpenCode's HTTP and MCP requests (#61582)
This commit is contained in:
@@ -7,9 +7,9 @@ import { rmSync } from "fs"
|
||||
import os from "os"
|
||||
import path from "path"
|
||||
import { Effect, Layer, Logger, ManagedRuntime, Scope } from "effect"
|
||||
import { FetchHttpClient } from "effect/http"
|
||||
import { Global } from "@opencode/util/global"
|
||||
import { LayerNode } from "@opencode/util/effect/layer-node"
|
||||
import { LayerNodePlatform } from "@opencode/util/effect/app-node-platform"
|
||||
import { AppNodeBuilder } from "../src/effect/app-node-builder"
|
||||
import { Bus } from "../src/bus"
|
||||
import { Database } from "../src/database/database"
|
||||
@@ -30,7 +30,12 @@ import { providerHost, host, integrationHost } from "../test/plugin/host"
|
||||
const root = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "opencode-location-bench-")))
|
||||
process.on("exit", () => rmSync(root, { recursive: true, force: true }))
|
||||
|
||||
const offline = Object.assign(() => Promise.reject(new Error("no network in the location benchmark")), {
|
||||
preconnect: () => {},
|
||||
})
|
||||
|
||||
const replacements = [
|
||||
LayerNodePlatform.httpFetch.replace(LayerNodePlatform.fetchLayer(offline)),
|
||||
Global.node.replace(
|
||||
Global.layerWith({
|
||||
home: path.join(root, "home"),
|
||||
@@ -55,12 +60,8 @@ const runtime = ManagedRuntime.make(
|
||||
).pipe(Layer.provide(Logger.layer([]))),
|
||||
)
|
||||
|
||||
const offline = Object.assign(() => Promise.reject(new Error("no network in the location benchmark")), {
|
||||
preconnect: () => {},
|
||||
})
|
||||
|
||||
const run = <A, E>(effect: Effect.Effect<A, E, LocationServiceMap.Service | ModelsDev.Service>) =>
|
||||
runtime.runPromise(effect.pipe(Effect.provideService(FetchHttpClient.Fetch, offline)))
|
||||
runtime.runPromise(effect)
|
||||
|
||||
// The variant's own checkout: a real project with its config, git and instructions.
|
||||
const ref = Location.Ref.make({ directory: AbsolutePath.make(process.cwd()) })
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
export * as AISDK from "./aisdk.js"
|
||||
|
||||
import { makeLocationNode } from "@opencode/util/effect/app-node"
|
||||
import { httpFetch } from "@opencode/util/effect/app-node-platform"
|
||||
import { APICallError } from "@ai-sdk/provider"
|
||||
import type {
|
||||
JSONSchema7,
|
||||
@@ -35,7 +36,7 @@ import { Auth, Endpoint, ProviderShared, RequestExecutor, type AnyRoute, type Ht
|
||||
import { isRecord } from "@opencode/util/record"
|
||||
import { Cause, Context, Effect, Layer, Option, Schema, Scope, Stream, Struct } from "effect"
|
||||
import { makeParser } from "effect/encoding/Sse"
|
||||
import { HttpClientRequest, HttpClientResponse } from "effect/http"
|
||||
import { FetchHttpClient, HttpClientRequest, HttpClientResponse } from "effect/http"
|
||||
import { AsyncLocalStorage } from "node:async_hooks"
|
||||
import type { ID, RuntimeInfo } from "./model.js"
|
||||
import { Provider } from "./provider.js"
|
||||
@@ -120,7 +121,8 @@ function wrapSSE(res: Response, ms: number, ctl: AbortController) {
|
||||
})
|
||||
}
|
||||
|
||||
function prepareOptions(model: RuntimeInfo, pkg: string) {
|
||||
// `outbound` sends requests for providers whose settings name no `fetch` of their own.
|
||||
function prepareOptions(model: RuntimeInfo, pkg: string, outbound: Fetch) {
|
||||
const projected = mapBodyToProviderOptions(model, pkg)
|
||||
const settings = model.settings ?? {}
|
||||
const options: Record<string, unknown> = {
|
||||
@@ -158,7 +160,7 @@ function prepareOptions(model: RuntimeInfo, pkg: string) {
|
||||
}
|
||||
}
|
||||
|
||||
const send: Fetch = isFetch(customFetch) ? customFetch : fetch
|
||||
const send: Fetch = isFetch(customFetch) ? customFetch : outbound
|
||||
const middleware = httpMiddleware.getStore()
|
||||
const res = await (
|
||||
middleware
|
||||
@@ -255,6 +257,7 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/AI
|
||||
export const locationLayer = Layer.effect(
|
||||
Service,
|
||||
Effect.gen(function* () {
|
||||
const outbound = yield* FetchHttpClient.Fetch
|
||||
let sdkHooks: ((event: SDKEvent) => Effect.Effect<void> | void)[] = []
|
||||
let languageHooks: ((event: LanguageEvent) => Effect.Effect<void> | void)[] = []
|
||||
const languages = new Map<string, LanguageModelV3>()
|
||||
@@ -333,7 +336,7 @@ export const locationLayer = Layer.effect(
|
||||
})
|
||||
|
||||
const packageName = Provider.packageName(model.package)
|
||||
const options = prepareOptions(model, packageName)
|
||||
const options = prepareOptions(model, packageName, outbound)
|
||||
const sdkKey = cacheKey({
|
||||
providerID: model.providerID,
|
||||
canonical: model.canonical,
|
||||
@@ -1042,4 +1045,4 @@ function providerErrorMessage(error: APICallError) {
|
||||
return error.message.trim() !== "" ? error.message : (message ?? (code === undefined ? prefix : `${prefix}: ${code}`))
|
||||
}
|
||||
|
||||
export const node = makeLocationNode({ service: Service, layer: locationLayer, deps: [] })
|
||||
export const node = makeLocationNode({ service: Service, layer: locationLayer, deps: [httpFetch] })
|
||||
|
||||
@@ -3,6 +3,7 @@ export * as McpConnection from "./connection.js"
|
||||
import type { OAuthClientProvider } from "@modelcontextprotocol/client"
|
||||
import type { Mcp } from "@opencode/schema/mcp"
|
||||
import { Cause, Effect, Exit, Scope } from "effect"
|
||||
import { FetchHttpClient } from "effect/http"
|
||||
import { Environment } from "../environment/index.js"
|
||||
import type { McpClient } from "./client.js"
|
||||
|
||||
@@ -39,6 +40,8 @@ export interface Input {
|
||||
readonly elicitation: McpClient.ElicitationHandler
|
||||
/** A stdio server is spawned on this execution plane. */
|
||||
readonly environment: Environment.Interface
|
||||
/** A remote server is reached with this fetch. */
|
||||
readonly fetch: typeof globalThis.fetch
|
||||
/** Each attempt's scope is forked from this one. */
|
||||
readonly scope: Scope.Scope
|
||||
/** Serializes lifecycle work. Callbacks and recovery take it; `start` and `stop` expect the caller to hold it. */
|
||||
@@ -165,6 +168,7 @@ export function make(input: Input): Connection {
|
||||
).pipe(
|
||||
Effect.flatMap((client) => client.tools().pipe(Effect.map((tools) => ({ client, tools })))),
|
||||
Effect.provideService(Environment.Service, input.environment),
|
||||
Effect.provideService(FetchHttpClient.Fetch, input.fetch),
|
||||
Scope.provide(scope),
|
||||
input.handshake,
|
||||
Effect.exit,
|
||||
|
||||
@@ -8,7 +8,9 @@ import type { Transport } from "@modelcontextprotocol/client"
|
||||
import { createHash } from "node:crypto"
|
||||
import { isDeepStrictEqual } from "node:util"
|
||||
import { Clock, Context, Effect, Exit, FiberSet, Latch, Layer, Schema, Scope, Semaphore, Stream, Types } from "effect"
|
||||
import { FetchHttpClient } from "effect/http"
|
||||
import { makeLocationNode } from "@opencode/util/effect/app-node"
|
||||
import { httpFetch } from "@opencode/util/effect/app-node-platform"
|
||||
import { Credential } from "../credential.js"
|
||||
import { Bus } from "../bus.js"
|
||||
import { Environment } from "../environment/index.js"
|
||||
@@ -189,6 +191,7 @@ export const layer = (options?: Options) =>
|
||||
const forms = yield* Form.Service
|
||||
const integration = yield* Integration.Service
|
||||
const credentials = yield* Credential.Service
|
||||
const fetch = yield* FetchHttpClient.Fetch
|
||||
const root = yield* Effect.scope
|
||||
const fork = yield* FiberSet.makeRuntime<never, void, never>()
|
||||
// Server scopes close in parallel on shutdown, so one slow server's stop grace does not delay
|
||||
@@ -245,6 +248,7 @@ export const layer = (options?: Options) =>
|
||||
const { McpOAuth } = yield* Effect.promise(() => import("./oauth.js"))
|
||||
return yield* McpOAuth.authorize({ name, config: remote, integrationID, methodID }).pipe(
|
||||
Effect.provideService(Credential.Service, credentials),
|
||||
Effect.provideService(FetchHttpClient.Fetch, fetch),
|
||||
)
|
||||
}),
|
||||
})
|
||||
@@ -267,6 +271,7 @@ export const layer = (options?: Options) =>
|
||||
const { McpOAuth } = yield* Effect.promise(() => import("./oauth.js"))
|
||||
return yield* McpOAuth.connectProvider({ config: entry.config, integrationID: entry.integrationID }).pipe(
|
||||
Effect.provideService(Credential.Service, credentials),
|
||||
Effect.provideService(FetchHttpClient.Fetch, fetch),
|
||||
)
|
||||
})
|
||||
|
||||
@@ -357,6 +362,7 @@ export const layer = (options?: Options) =>
|
||||
authProvider: Effect.suspend(() => connectProvider(entry)),
|
||||
elicitation,
|
||||
environment,
|
||||
fetch,
|
||||
scope: servers,
|
||||
lock: locks.withLock(name),
|
||||
handshake: (effect) => (config.type === "remote" ? endpointLoads.withLock(config.url)(effect) : effect),
|
||||
@@ -856,7 +862,7 @@ export function configured(options?: Options) {
|
||||
return makeLocationNode({
|
||||
service: Service,
|
||||
layer: layer(options),
|
||||
deps: [Location.node, Environment.node, Bus.node, Form.node, Integration.node, Credential.node],
|
||||
deps: [Location.node, Environment.node, Bus.node, Form.node, Integration.node, Credential.node, httpFetch],
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -17,10 +17,10 @@ import {
|
||||
} from "@modelcontextprotocol/client"
|
||||
import { OAuthMetadataSchema, OpenIdProviderDiscoveryMetadataSchema } from "@modelcontextprotocol/core"
|
||||
import { Cause, Deferred, Effect } from "effect"
|
||||
import { FetchHttpClient } from "effect/http"
|
||||
import type { ServerResponse } from "node:http"
|
||||
import type { AddressInfo } from "node:net"
|
||||
import { ConfigMCP } from "@opencode/schema/config/mcp"
|
||||
import { noTimeoutFetch } from "@opencode/util/no-timeout-fetch"
|
||||
import { Credential } from "../credential.js"
|
||||
import { OauthCallbackPage } from "../oauth/page.js"
|
||||
import type { Integration } from "../integration.js"
|
||||
@@ -40,24 +40,30 @@ const refreshKey = (url: string | URL, init: RequestInit | undefined) => {
|
||||
// Bun's and Node's fetch types both apply here and the SDK's FetchLike wants Bun's Response, so the
|
||||
// overload is picked by annotation and clone() is pinned to the type it was called on. A tool call
|
||||
// can stay silent for minutes, so Bun's idle deadline is off; the MCP execution timeout still applies.
|
||||
const base: FetchLike = noTimeoutFetch
|
||||
const share = (pending: ReturnType<FetchLike>) => pending.then((response) => response.clone() as typeof response)
|
||||
|
||||
const send: FetchLike = (url, init) => {
|
||||
const key = refreshKey(url, init)
|
||||
if (key === undefined) return base(url, init)
|
||||
const current = refreshes.get(key)
|
||||
if (current) return share(current)
|
||||
const pending = base(url, init).finally(() => {
|
||||
if (refreshes.get(key) === pending) refreshes.delete(key)
|
||||
})
|
||||
refreshes.set(key, pending)
|
||||
return share(pending)
|
||||
}
|
||||
// Requests use `FetchHttpClient.Fetch`, which an embedder can replace, rather than the global fetch.
|
||||
const sender = Effect.gen(function* () {
|
||||
const fetch = yield* FetchHttpClient.Fetch
|
||||
const base: FetchLike = (url, init) => fetch(url, { ...init, timeout: false } as BunFetchRequestInit)
|
||||
const send: FetchLike = (url, init) => {
|
||||
const key = refreshKey(url, init)
|
||||
if (key === undefined) return base(url, init)
|
||||
const current = refreshes.get(key)
|
||||
if (current) return share(current)
|
||||
const pending = base(url, init).finally(() => {
|
||||
if (refreshes.get(key) === pending) refreshes.delete(key)
|
||||
})
|
||||
refreshes.set(key, pending)
|
||||
return share(pending)
|
||||
}
|
||||
return send
|
||||
})
|
||||
|
||||
export const loggedFetch = (fields: { readonly server: string; readonly directory?: string }) =>
|
||||
Effect.gen(function* () {
|
||||
const run = Effect.runPromiseWith(yield* Effect.context())
|
||||
const send = yield* sender
|
||||
const request: FetchLike = (url, init) => {
|
||||
const grant = init?.body instanceof URLSearchParams ? init.body.get("grant_type") : undefined
|
||||
const operation = grant === "refresh_token" ? "refresh" : grant === "authorization_code" ? "exchange" : undefined
|
||||
@@ -136,6 +142,7 @@ interface Store {
|
||||
interface Options {
|
||||
readonly config: typeof ConfigMCP.Remote.Type
|
||||
readonly store: Store
|
||||
readonly fetch: FetchLike
|
||||
/** Absent on connect: the provider then refuses to register a client or redirect, ending in needs_auth. */
|
||||
readonly redirect?: {
|
||||
readonly url: string
|
||||
@@ -160,7 +167,7 @@ const provider = (options: Options): OAuthClientProvider => {
|
||||
return {
|
||||
redirectUrl,
|
||||
discoveryState: async () => {
|
||||
discovery ??= await configuredDiscovery({ config: options.config, fetchFn: send })
|
||||
discovery ??= await configuredDiscovery({ config: options.config, fetchFn: options.fetch })
|
||||
return discovery
|
||||
},
|
||||
saveDiscoveryState: (state) => {
|
||||
@@ -203,7 +210,9 @@ const provider = (options: Options): OAuthClientProvider => {
|
||||
redirectToAuthorization: (url) => {
|
||||
if (!redirect) throw refuse("user authorization")
|
||||
if (url.protocol !== "http:" && url.protocol !== "https:")
|
||||
throw new Error(`MCP server "${options.config.url}" returned a ${url.protocol} authorization URL; only http and https are supported`)
|
||||
throw new Error(
|
||||
`MCP server "${options.config.url}" returned a ${url.protocol} authorization URL; only http and https are supported`,
|
||||
)
|
||||
return redirect.open(url)
|
||||
},
|
||||
...(options.invalidate ? { invalidateCredentials: options.invalidate } : {}),
|
||||
@@ -279,8 +288,9 @@ export const connectProvider = Effect.fnUntraced(function* (input: {
|
||||
}) {
|
||||
const credentials = yield* Credential.Service
|
||||
const run = Effect.runPromiseWith(yield* Effect.context())
|
||||
const fetch = yield* sender
|
||||
const found = (yield* credentials.list(input.integrationID)).at(-1)
|
||||
if (!found || found.value.type !== "oauth") return provider({ config: input.config, store: memoryStore() })
|
||||
if (!found || found.value.type !== "oauth") return provider({ config: input.config, store: memoryStore(), fetch })
|
||||
const id = found.id
|
||||
const methodID = found.value.methodID
|
||||
const read = async () => {
|
||||
@@ -291,6 +301,7 @@ export const connectProvider = Effect.fnUntraced(function* (input: {
|
||||
let presented = found.value.refresh
|
||||
return provider({
|
||||
config: input.config,
|
||||
fetch,
|
||||
invalidate: async (scope) => {
|
||||
if (scope === "verifier" || scope === "discovery") return
|
||||
const oauth = await read()
|
||||
@@ -389,7 +400,9 @@ export const authorize = (input: {
|
||||
if (!value) return fail("Missing authorization code", "missing_code")
|
||||
// The page waits for the token exchange so the browser never reports success for a rejected code.
|
||||
if (
|
||||
!Effect.runSync(Deferred.succeed(code, { code: value, iss: url.searchParams.get("iss") ?? undefined, response }))
|
||||
!Effect.runSync(
|
||||
Deferred.succeed(code, { code: value, iss: url.searchParams.get("iss") ?? undefined, response }),
|
||||
)
|
||||
)
|
||||
response.writeHead(409).end("OAuth callback already received")
|
||||
})
|
||||
@@ -450,6 +463,7 @@ export const authorize = (input: {
|
||||
const oauthProvider = provider({
|
||||
config: input.config,
|
||||
store,
|
||||
fetch: fetchFn,
|
||||
clientMetadataUrl: cimd ? CLIENT_METADATA_URL : undefined,
|
||||
discovery: { ...discovery, resourceMetadataUrl: resourceMetadataUrl?.toString() },
|
||||
redirect: {
|
||||
|
||||
@@ -4,10 +4,10 @@ import { LLMClient } from "@opencode/ai/effect"
|
||||
import type { Plugin } from "@opencode/plugin/effect/plugin"
|
||||
import { LayerNode } from "@opencode/util/effect/layer-node"
|
||||
import type { Generation } from "./service.js"
|
||||
import { httpClient } from "@opencode/util/effect/app-node-platform"
|
||||
import { httpClient, httpFetch } from "@opencode/util/effect/app-node-platform"
|
||||
import { AppProcess } from "@opencode/util/process"
|
||||
import { Context, Effect, Scope } from "effect"
|
||||
import { HttpClient } from "effect/http"
|
||||
import { FetchHttpClient, HttpClient } from "effect/http"
|
||||
import { Agent } from "../agent.js"
|
||||
import { Artifact } from "../artifact.js"
|
||||
import { Model } from "../model.js"
|
||||
@@ -118,6 +118,7 @@ import { WellKnownPlugin } from "../wellknown/plugin.js"
|
||||
|
||||
const services = [
|
||||
Agent.Service,
|
||||
FetchHttpClient.Fetch,
|
||||
Artifact.Service,
|
||||
AppProcess.Service,
|
||||
Provider.Service,
|
||||
@@ -195,6 +196,7 @@ export const requirements = LayerNode.group([
|
||||
FSUtil.node,
|
||||
Global.node,
|
||||
httpClient,
|
||||
httpFetch,
|
||||
Image.node,
|
||||
InstructionDiscovery.node,
|
||||
Integration.node,
|
||||
|
||||
@@ -2,7 +2,7 @@ import type { IntegrationOAuthMethodRegistration } from "@opencode/plugin/effect
|
||||
import type { SessionRequestKind, SessionTitle } from "@opencode/plugin/effect/session"
|
||||
import { isRecord } from "@opencode/util/record"
|
||||
import { Effect, Option, Schema } from "effect"
|
||||
import { HttpClient } from "effect/http"
|
||||
import { FetchHttpClient, HttpClient } from "effect/http"
|
||||
import { IntegrationConnection } from "../../integration/connection.js"
|
||||
import { Credential } from "../../credential.js"
|
||||
import { CopilotModels } from "../../github-copilot/models.js"
|
||||
@@ -166,6 +166,7 @@ export const GithubCopilotPlugin = define({
|
||||
const providers = yield* Provider.Service
|
||||
const cache = yield* SharedCache.Service
|
||||
const http = yield* HttpClient.HttpClient
|
||||
const outbound = yield* FetchHttpClient.Fetch
|
||||
const discovered = yield* ProviderRefresh.start({
|
||||
integrationID: "github-copilot",
|
||||
initial: {} as {
|
||||
@@ -293,7 +294,7 @@ export const GithubCopilotPlugin = define({
|
||||
const model = utilityTitleModels.find((id) => loaded.models?.has(id))
|
||||
if (!model) return
|
||||
evt.result = yield* utilityTitle(
|
||||
{ baseURL: loaded.baseURL, token: loaded.token, model, app: ctx.app },
|
||||
{ baseURL: loaded.baseURL, token: loaded.token, model, app: ctx.app, fetch: outbound },
|
||||
evt,
|
||||
).pipe(
|
||||
Effect.tapError((cause) => Effect.logDebug("Copilot utility title failed", { model, cause })),
|
||||
|
||||
@@ -26,6 +26,8 @@ import { compileRequest } from "@opencode/ai/testing"
|
||||
import { expect } from "bun:test"
|
||||
import { Effect, Layer } from "effect"
|
||||
import { HttpClientRequest, HttpClientResponse } from "effect/http"
|
||||
import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder"
|
||||
import { LayerNodePlatform } from "@opencode/util/effect/app-node-platform"
|
||||
import { testEffect } from "./lib/effect"
|
||||
|
||||
const it = testEffect(AISDK.locationLayer)
|
||||
@@ -566,11 +568,19 @@ it.effect("normalizes file data across AI SDK prompt parts", () =>
|
||||
Message.user([
|
||||
{ type: "media", media: Media.bytes(bytes, "image/png"), filename: "bytes.png" },
|
||||
{ type: "media", media: Media.base64("AAAA", "image/png"), filename: "base64.png" },
|
||||
{ type: "media", media: Media.fromDataUrl("data:image/png;charset=utf-8;base64,AQID"), filename: "inline.png" },
|
||||
{
|
||||
type: "media",
|
||||
media: Media.fromDataUrl("data:image/png;charset=utf-8;base64,AQID"),
|
||||
filename: "inline.png",
|
||||
},
|
||||
{ type: "media", media: Media.url("https://example.com/image.png", { mediaType: "image/png" }) },
|
||||
{ type: "media", media: Media.base64("s3://bucket/image.png", "image/png") },
|
||||
]),
|
||||
Message.assistant({ type: "media", media: Media.url("http://example.com/document.pdf", { mediaType: "application/pdf" }), filename: "document.pdf" }),
|
||||
Message.assistant({
|
||||
type: "media",
|
||||
media: Media.url("http://example.com/document.pdf", { mediaType: "application/pdf" }),
|
||||
filename: "document.pdf",
|
||||
}),
|
||||
Message.tool({
|
||||
id: "call_1",
|
||||
name: "screenshot",
|
||||
@@ -713,6 +723,38 @@ const compatibleModel = Effect.fn(function* (customFetch: typeof fetch) {
|
||||
)
|
||||
})
|
||||
|
||||
// Not `it`, whose AISDK layer is already built with the harness's fetch and would be reused here.
|
||||
testEffect(Layer.empty).effect("sends AI SDK requests through the embedder's fetch when the provider names none", () =>
|
||||
Effect.gen(function* () {
|
||||
const sent: string[] = []
|
||||
const embedder = async (input: Request | string | URL) => {
|
||||
sent.push(input instanceof Request ? input.url : String(input))
|
||||
return new Response(chatChunk("embedded"), { headers: { "content-type": "text/event-stream" } })
|
||||
}
|
||||
const response = yield* Effect.gen(function* () {
|
||||
const aisdk = yield* AISDK.Service
|
||||
yield* aisdk.hook.sdk((event) => {
|
||||
event.sdk = createOpenAICompatible({
|
||||
...event.options,
|
||||
name: String(event.options.name),
|
||||
baseURL: String(event.options.baseURL),
|
||||
})
|
||||
})
|
||||
const resolved = yield* aisdk.model(
|
||||
model("@ai-sdk/openai-compatible", { apiKey: "test", baseURL: "https://example.test/v1" }),
|
||||
)
|
||||
return yield* LLMClient.generate(ai.request({ model: resolved, prompt: "Hello" })).pipe(Effect.provide(client))
|
||||
}).pipe(
|
||||
Effect.provide(
|
||||
AppNodeBuilder.build(AISDK.node, [LayerNodePlatform.httpFetch.replace(LayerNodePlatform.fetchLayer(embedder))]),
|
||||
),
|
||||
)
|
||||
|
||||
expect(sent).toEqual(["https://example.test/v1/chat/completions"])
|
||||
expect(response.text).toBe("embedded")
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("routes AI SDK requests and responses through HTTP hook middleware", () =>
|
||||
Effect.gen(function* () {
|
||||
const sent: Array<{ url: string; headers: Headers; body: string }> = []
|
||||
@@ -753,12 +795,7 @@ it.effect("routes AI SDK requests and responses through HTTP hook middleware", (
|
||||
expect(sent[0]?.headers.get("x-hook")).toBe("applied")
|
||||
expect(sent[0]?.headers.get("authorization")).toBe("Bearer test")
|
||||
expect(JSON.parse(sent[0]?.body ?? "")).toMatchObject({ model: "api-model" })
|
||||
expect(seen).toEqual([
|
||||
"POST https://example.test/v1/chat/completions",
|
||||
sent[0]?.body,
|
||||
sent[0]?.body,
|
||||
"status 200",
|
||||
])
|
||||
expect(seen).toEqual(["POST https://example.test/v1/chat/completions", sent[0]?.body, sent[0]?.body, "status 200"])
|
||||
expect(response.events.filter(LLMEvent.is.textDelta).map((event) => event.text)).toEqual(["rewritten"])
|
||||
}),
|
||||
)
|
||||
|
||||
@@ -0,0 +1,14 @@
|
||||
import { createMcpHandler, Server } from "@modelcontextprotocol/server"
|
||||
|
||||
/** A web-standard MCP server with one `echo` tool, for tests that route MCP HTTP through their own fetch. */
|
||||
export const echoMcpHandler = () =>
|
||||
createMcpHandler(() => {
|
||||
const protocol = new Server({ name: "echo", version: "1.0.0" }, { capabilities: { tools: {} } })
|
||||
protocol.setRequestHandler("tools/list", () =>
|
||||
Promise.resolve({ tools: [{ name: "echo", inputSchema: { type: "object" as const, properties: {} } }] }),
|
||||
)
|
||||
protocol.setRequestHandler("tools/call", (request) =>
|
||||
Promise.resolve({ content: [{ type: "text" as const, text: `called ${request.params.name}` }] }),
|
||||
)
|
||||
return protocol
|
||||
})
|
||||
@@ -10,10 +10,10 @@ const body = <A, E, R>(value: Body<A, E, R>) => Effect.suspend(() => (typeof val
|
||||
const loopback = new Set(["127.0.0.1", "localhost", "[::1]"])
|
||||
|
||||
// Core is env-free, so nothing tells the default node graph to stay offline;
|
||||
// a test that boots it would phone home through FetchHttpClient. Every
|
||||
// FetchHttpClient reads this reference at request time, so refusing here covers
|
||||
// each node that shares the default client, while an explicit HttpClient
|
||||
// replacement never reaches it. Callers such as ModelsDev swallow request
|
||||
// a test that boots it would phone home through FetchHttpClient. The default
|
||||
// client node captures this reference when the test builds its layer, so
|
||||
// refusing here covers each node that shares the default client, while an
|
||||
// explicit HttpClient or fetch replacement never reaches it. Callers such as ModelsDev swallow request
|
||||
// failures, so the harness also records the attempt and fails the test itself.
|
||||
export const refuseNetwork = (violations: string[]): typeof fetch =>
|
||||
Object.assign(
|
||||
|
||||
@@ -5,7 +5,7 @@ import { Session } from "@opencode/core/session"
|
||||
import { Location } from "@opencode/core/location"
|
||||
import { describe, expect, test } from "bun:test"
|
||||
import { Effect } from "effect"
|
||||
import { HttpClient, HttpClientResponse } from "effect/http"
|
||||
import { FetchHttpClient, HttpClient, HttpClientResponse } from "effect/http"
|
||||
import { TestClock } from "effect/testing"
|
||||
import { Model } from "@opencode/core/model"
|
||||
import { ModelResolver } from "@opencode/core/model-resolver"
|
||||
@@ -348,6 +348,69 @@ describe("GithubCopilotPlugin", () => {
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("sends the utility title request through the host's fetch", () =>
|
||||
Effect.gen(function* () {
|
||||
const credentials = yield* Credential.Service
|
||||
const providers = yield* Provider.Service
|
||||
const hooks = yield* PluginHooks.Service
|
||||
const http = HttpClient.make((request) =>
|
||||
Effect.succeed(
|
||||
HttpClientResponse.fromWeb(
|
||||
request,
|
||||
Response.json({
|
||||
data: [
|
||||
{
|
||||
model_picker_enabled: true,
|
||||
id: "gpt-4o-mini",
|
||||
name: "GPT-4o mini",
|
||||
version: "gpt-4o-mini",
|
||||
capabilities: {
|
||||
family: "gpt-4o-mini",
|
||||
limits: { max_output_tokens: 1, max_prompt_tokens: 1 },
|
||||
supports: { tool_calls: true },
|
||||
},
|
||||
},
|
||||
],
|
||||
}),
|
||||
),
|
||||
),
|
||||
)
|
||||
const sent: string[] = []
|
||||
const outbound = Object.assign(
|
||||
async (input: Parameters<typeof fetch>[0]) => {
|
||||
sent.push(String(input))
|
||||
return Response.json({ choices: [{ message: { content: "Faster Python" } }] })
|
||||
},
|
||||
{ preconnect: fetch.preconnect },
|
||||
)
|
||||
yield* providers.transform((draft) => {
|
||||
draft.update(Provider.ID.githubCopilot, () => {})
|
||||
})
|
||||
yield* credentials.create({
|
||||
integrationID: Integration.ID.make("github-copilot"),
|
||||
value: Credential.OAuth.make({
|
||||
type: "oauth",
|
||||
methodID: Integration.MethodID.make("device"),
|
||||
access: "token",
|
||||
refresh: "token",
|
||||
expires: 0,
|
||||
}),
|
||||
})
|
||||
yield* addPlugin().pipe(
|
||||
Effect.provideService(HttpClient.HttpClient, http),
|
||||
Effect.provideService(FetchHttpClient.Fetch, outbound),
|
||||
)
|
||||
yield* drain
|
||||
|
||||
const event = yield* hooks.trigger("session", "title", {
|
||||
...titleRequest,
|
||||
model: Model.Ref.make({ providerID: Provider.ID.githubCopilot, id: Model.ID.make("gpt-5") }),
|
||||
})
|
||||
expect(event.result).toBe("Faster Python")
|
||||
expect(sent).toEqual(["https://api.githubcopilot.com/chat/completions"])
|
||||
}),
|
||||
)
|
||||
|
||||
it.live("fails the utility title on rate limits so the billable path can run", () =>
|
||||
Effect.gen(function* () {
|
||||
const error = yield* utilityTitle(
|
||||
|
||||
@@ -2,6 +2,7 @@ import { Database } from "@opencode/core/database/database"
|
||||
import { ModelsDev } from "@opencode/core/models-dev"
|
||||
import { PersistentPty } from "@opencode/core/persistent-pty"
|
||||
import { Schema } from "effect"
|
||||
import type { LayerNodePlatform } from "@opencode/util/effect/app-node-platform"
|
||||
|
||||
export const ServerOptions = Schema.Struct({
|
||||
app: Schema.optional(
|
||||
@@ -24,6 +25,12 @@ export const ServerOptions = Schema.Struct({
|
||||
}),
|
||||
),
|
||||
models: Schema.optional(ModelsDev.Options),
|
||||
/**
|
||||
* Replaces `globalThis.fetch` for OpenCode's own HTTP requests: OpenCode Console config, model requests, web fetch
|
||||
* and search, well-known config, and remote MCP servers. Agent code in Code Mode keeps `globalThis.fetch`. For
|
||||
* example, a Cloudflare Worker can route requests through a Service Binding.
|
||||
*/
|
||||
fetch: Schema.optional(Schema.declare((input): input is LayerNodePlatform.Fetch => typeof input === "function")),
|
||||
config: Schema.optional(
|
||||
Schema.Struct({
|
||||
directory: Schema.optional(Schema.String),
|
||||
|
||||
@@ -3,7 +3,7 @@ import { V1Migration } from "@opencode/core/database/v1-migration"
|
||||
import { App } from "@opencode/core/app"
|
||||
import { LayerNode } from "@opencode/util/effect/layer-node"
|
||||
import { Node } from "@opencode/util/effect/app-node"
|
||||
import { httpClient } from "@opencode/util/effect/app-node-platform"
|
||||
import { fetchLayer, httpClient, httpFetch } from "@opencode/util/effect/app-node-platform"
|
||||
import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder"
|
||||
import { Bus } from "@opencode/core/bus"
|
||||
import { EventLogger } from "@opencode/core/event-logger"
|
||||
@@ -127,6 +127,7 @@ function makeRoutes<AuthError, AuthServices>(
|
||||
Bus.node.replace(Bus.configured({ persist: options.events?.persist })),
|
||||
App.node.replace(App.configured(options.app)),
|
||||
ModelsDev.node.replace(ModelsDev.configured(options.models)),
|
||||
httpFetch.replace(fetchLayer(options.fetch)),
|
||||
Watcher.node.replace(Watcher.configured({ enabled: options.fs?.filewatcher })),
|
||||
FileWatch.node.replace(FileWatch.configured({ enabled: options.fs?.filewatcher })),
|
||||
FileSystemSearch.node.replace(FileSystemSearch.configured({ fff: options.fs?.fff })),
|
||||
|
||||
@@ -46,6 +46,8 @@ export interface Options {
|
||||
readonly config?: { readonly content?: string }
|
||||
/** models.dev catalog options; the bundled snapshot is the boot-time floor either way. */
|
||||
readonly models?: ServerOptions["models"]
|
||||
/** Replaces `globalThis.fetch` for OpenCode's own HTTP requests, e.g. to route them through a Service Binding. */
|
||||
readonly fetch?: ServerOptions["fetch"]
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -71,6 +73,7 @@ export function serverOptions(options: Options): ServerOptions {
|
||||
events: { persist: true },
|
||||
config: { content: options.config?.content },
|
||||
models: options.models,
|
||||
fetch: options.fetch,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
import { expect } from "bun:test"
|
||||
import { Location } from "@opencode/core/location"
|
||||
import { LocationServiceMap } from "@opencode/core/location-service-map"
|
||||
import { Mcp } from "@opencode/core/mcp/index"
|
||||
import { Context, Effect, Layer } from "effect"
|
||||
import { HttpEffect, HttpRouter, HttpServer } from "effect/http"
|
||||
import { echoMcpHandler } from "../../core/test/fixture/mcp-echo"
|
||||
import { waitUntil } from "../../core/test/lib/eventually"
|
||||
import { tmpdirScoped } from "../../core/test/fixture/tmpdir"
|
||||
import { it } from "../../core/test/lib/effect"
|
||||
import { createRoutes } from "../src/routes"
|
||||
|
||||
// `.invalid` never resolves, so only the embedder's fetch can answer these.
|
||||
const consoleServer = "https://console.invalid"
|
||||
const mcpServer = "https://mcp.invalid/mcp"
|
||||
|
||||
it.live("sends Console config and remote MCP requests through the embedder's fetch", () =>
|
||||
Effect.gen(function* () {
|
||||
const tmp = yield* tmpdirScoped("opencode-fetch-option-")
|
||||
const requested: string[] = []
|
||||
const methods: string[] = []
|
||||
const mcp = echoMcpHandler()
|
||||
yield* Effect.addFinalizer(() => Effect.promise(() => mcp.close()))
|
||||
const fetch = async (input: Request | string | URL, init?: RequestInit) => {
|
||||
const request = new Request(input, init)
|
||||
const url = new URL(request.url)
|
||||
requested.push(`${url.origin}${url.pathname}`)
|
||||
if (url.pathname === "/api/v2/config") return Response.json({})
|
||||
if (`${url.origin}${url.pathname}` !== mcpServer) return new Response("Not found", { status: 404 })
|
||||
const body: unknown = request.method === "POST" ? await request.clone().json() : undefined
|
||||
if (typeof body === "object" && body !== null && "method" in body && typeof body.method === "string")
|
||||
methods.push(body.method)
|
||||
return mcp.fetch(request)
|
||||
}
|
||||
const context = yield* Layer.build(
|
||||
createRoutes({
|
||||
password: "secret",
|
||||
database: { path: ":memory:" },
|
||||
models: { fetch: false },
|
||||
fs: { filewatcher: false },
|
||||
fetch,
|
||||
config: {
|
||||
directory: tmp.path,
|
||||
project: false,
|
||||
content: JSON.stringify({ mcp: { servers: { gateway: { type: "remote", url: mcpServer, oauth: false } } } }),
|
||||
},
|
||||
}).pipe(Layer.provide(HttpServer.layerServices)),
|
||||
)
|
||||
const handler = Context.get(context, HttpRouter.HttpRouter)
|
||||
.asHttpEffect()
|
||||
.pipe(HttpEffect.toWebHandlerWith(context))
|
||||
const request = (method: "GET" | "POST", route: string, body?: unknown) =>
|
||||
Effect.promise(async (signal) => {
|
||||
const url = new URL(route, "http://opencode.local")
|
||||
url.searchParams.set("location[directory]", tmp.path)
|
||||
const response = await handler(
|
||||
new Request(url, {
|
||||
method,
|
||||
headers: {
|
||||
authorization: `Basic ${btoa("opencode:secret")}`,
|
||||
...(body === undefined ? {} : { "content-type": "application/json" }),
|
||||
},
|
||||
body: body === undefined ? undefined : JSON.stringify(body),
|
||||
signal,
|
||||
}),
|
||||
)
|
||||
return response.status
|
||||
})
|
||||
const inLocation = <A, E>(effect: Effect.Effect<A, E, Mcp.Service>) =>
|
||||
effect.pipe(
|
||||
Effect.provide(
|
||||
LocationServiceMap.Service.get(
|
||||
LocationServiceMap.canonical(Location.Ref.make({ directory: tmp.path as Location.Ref["directory"] })),
|
||||
),
|
||||
),
|
||||
Effect.provideContext(context),
|
||||
)
|
||||
|
||||
expect(
|
||||
yield* request("POST", "/api/credential", {
|
||||
integrationID: "opencode",
|
||||
value: { type: "key", key: "secret", metadata: { server: consoleServer } },
|
||||
}),
|
||||
).toBe(200)
|
||||
expect(yield* request("GET", "/api/provider")).toBe(200)
|
||||
yield* waitUntil(Effect.sync(() => requested.includes(`${consoleServer}/api/v2/config`)))
|
||||
|
||||
yield* waitUntil(
|
||||
inLocation(Mcp.Service.use((service) => service.tools())).pipe(
|
||||
Effect.map((tools) => tools.some((tool) => tool.server === "gateway" && tool.name === "echo")),
|
||||
),
|
||||
)
|
||||
const result = yield* inLocation(
|
||||
Mcp.Service.use((service) => service.callTool({ server: "gateway", name: "echo" })),
|
||||
)
|
||||
expect(result.content).toEqual([{ type: "text", text: "called echo" }])
|
||||
expect(methods).toEqual(expect.arrayContaining(["initialize", "tools/list", "tools/call"]))
|
||||
|
||||
// A reconnect requested over the API starts the connection in the request's fiber rather than the MCP service's.
|
||||
methods.length = 0
|
||||
expect(yield* request("POST", "/api/experimental/mcp/gateway/connect")).toBe(204)
|
||||
expect(methods).toContain("initialize")
|
||||
}),
|
||||
)
|
||||
@@ -166,6 +166,13 @@ function occupy(port: number, cancel = false) {
|
||||
})
|
||||
}
|
||||
|
||||
// A server without its own config directory reads the developer's global config, whose remote MCP servers would
|
||||
// reach the network.
|
||||
const isolated = Effect.gen(function* () {
|
||||
const config = yield* Effect.acquireDisposable(Effect.promise(() => tmpdir("opencode-openai-oauth-")))
|
||||
return yield* ServerFetch.make({ ...options, config: { ...options.config, directory: config.path } })
|
||||
})
|
||||
|
||||
const connectOpenAI = (handler: Handler) =>
|
||||
Effect.gen(function* () {
|
||||
// Catalog endpoints no longer wait for plugin initialization.
|
||||
@@ -426,7 +433,7 @@ it.live("returns 404 when a previously readable file is deleted", () =>
|
||||
it.live("cancels a stale OpenAI OAuth callback server before falling back", () =>
|
||||
Effect.gen(function* () {
|
||||
const requests = yield* occupy(1455, true)
|
||||
const handler = yield* ServerFetch.make(options)
|
||||
const handler = yield* isolated
|
||||
const response = yield* connectOpenAI(handler)
|
||||
|
||||
expect(response.status).toBe(200)
|
||||
@@ -439,7 +446,7 @@ it.live("cancels a stale OpenAI OAuth callback server before falling back", () =
|
||||
it.live("falls back to port 1457 when OpenAI OAuth port 1455 remains busy", () =>
|
||||
Effect.gen(function* () {
|
||||
const requests = yield* occupy(1455)
|
||||
const handler = yield* ServerFetch.make(options)
|
||||
const handler = yield* isolated
|
||||
const response = yield* connectOpenAI(handler)
|
||||
|
||||
expect(response.status).toBe(200)
|
||||
@@ -455,7 +462,7 @@ it.live(
|
||||
Effect.gen(function* () {
|
||||
yield* occupy(1455)
|
||||
yield* occupy(1457)
|
||||
const handler = yield* ServerFetch.make(options)
|
||||
const handler = yield* isolated
|
||||
const response = yield* connectOpenAI(handler)
|
||||
|
||||
expect(response.status).toBe(400)
|
||||
|
||||
@@ -62,7 +62,12 @@ const serve = (database: string, mediaBlob: boolean) =>
|
||||
ServerFetch.make({
|
||||
app: { version: "test" },
|
||||
database: { path: database },
|
||||
config: { project: false, content: JSON.stringify(mediaBlob ? { experimental: { media_blob: true } } : {}) },
|
||||
// Its own global config directory, so the developer's global config can't add remote MCP servers.
|
||||
config: {
|
||||
directory: path.dirname(database),
|
||||
project: false,
|
||||
content: JSON.stringify(mediaBlob ? { experimental: { media_blob: true } } : {}),
|
||||
},
|
||||
models: { fetch: false },
|
||||
fs: { filewatcher: false },
|
||||
})
|
||||
|
||||
@@ -34,3 +34,9 @@ test("accepts an optional CORS allowlist", () => {
|
||||
])
|
||||
expect(Option.isNone(decode({ cors: "http://192.168.1.10:3001" }))).toBe(true)
|
||||
})
|
||||
|
||||
test("accepts an embedder fetch and rejects other values", () => {
|
||||
const fetch = () => Promise.resolve(new Response())
|
||||
expect(Option.getOrThrow(decode({ fetch })).fetch).toBe(fetch)
|
||||
expect(Option.isNone(decode({ fetch: "https://example.com" }))).toBe(true)
|
||||
})
|
||||
|
||||
@@ -2,12 +2,39 @@
|
||||
// ioredis, and node:sqlite, which runtimes such as workerd cannot load.
|
||||
import * as NodeFileSystem from "@effect/platform-node/NodeFileSystem"
|
||||
import * as NodePath from "@effect/platform-node/NodePath"
|
||||
import { FileSystem, Path } from "effect"
|
||||
import { Effect, FileSystem, Layer, Path } from "effect"
|
||||
import { FetchHttpClient, HttpClient } from "effect/http"
|
||||
import { makeGlobalNode } from "./app-node.js"
|
||||
|
||||
export const filesystem = makeGlobalNode({ service: FileSystem.FileSystem, layer: NodeFileSystem.layer, deps: [] })
|
||||
export const path = makeGlobalNode({ service: Path.Path, layer: NodePath.layer, deps: [] })
|
||||
export const httpClient = makeGlobalNode({ service: HttpClient.HttpClient, layer: FetchHttpClient.layer, deps: [] })
|
||||
// `RequestInfo | URL`, spelled out because not every runtime declares `RequestInfo` globally.
|
||||
export type Fetch = (input: Request | string | URL, init?: RequestInit) => Promise<Response>
|
||||
/**
|
||||
* The fetch behind `httpClient`, AI SDK providers and remote MCP servers: `fetch` when given, so an embedder can route requests through
|
||||
* e.g. a Service Binding, otherwise the `FetchHttpClient.Fetch` in scope when the graph is built.
|
||||
*/
|
||||
export const fetchLayer = (fetch?: Fetch) =>
|
||||
Layer.effect(
|
||||
FetchHttpClient.Fetch,
|
||||
Effect.gen(function* () {
|
||||
// Bun's `typeof fetch` adds `preconnect`, which nothing here calls.
|
||||
return fetch ? (fetch as typeof globalThis.fetch) : yield* FetchHttpClient.Fetch
|
||||
}),
|
||||
)
|
||||
export const httpFetch = makeGlobalNode({ service: FetchHttpClient.Fetch, layer: fetchLayer(), deps: [] })
|
||||
// Provided per request, so a caller's context can't swap `httpFetch` for another fetch.
|
||||
export const httpClient = makeGlobalNode({
|
||||
service: HttpClient.HttpClient,
|
||||
layer: Layer.effect(
|
||||
HttpClient.HttpClient,
|
||||
Effect.gen(function* () {
|
||||
const fetch = yield* FetchHttpClient.Fetch
|
||||
const client = yield* HttpClient.HttpClient.pipe(Effect.provide(FetchHttpClient.layer))
|
||||
return HttpClient.transformResponse(client, Effect.provideService(FetchHttpClient.Fetch, fetch))
|
||||
}),
|
||||
),
|
||||
deps: [httpFetch],
|
||||
})
|
||||
|
||||
export * as LayerNodePlatform from "./app-node-platform.js"
|
||||
|
||||
@@ -55,6 +55,22 @@ a Durable Object without running cleanup, so correctness does not depend on `clo
|
||||
}
|
||||
```
|
||||
|
||||
## Route outbound requests
|
||||
|
||||
Pass `fetch` to route OpenCode's outbound HTTP requests (OpenCode Console config, model requests, web fetch and
|
||||
search, well-known config, and remote MCP servers) through a custom handler such as a Service Binding. Agent code in
|
||||
Code Mode continues to use `globalThis.fetch`.
|
||||
|
||||
```ts
|
||||
OpenCodeWorkerd.create({
|
||||
storage: state.storage,
|
||||
fetch: (input, init) => {
|
||||
const request = new Request(input, init)
|
||||
return new URL(request.url).hostname === "console.example.com" ? env.CONSOLE.fetch(request) : fetch(request)
|
||||
},
|
||||
})
|
||||
```
|
||||
|
||||
## Customize
|
||||
|
||||
Customize your OpenCode instance by registering plugins bundled with your
|
||||
|
||||
Reference in New Issue
Block a user