mcp
This commit is contained in:
1 parent
9f578780bd
commit
5807132ac6
6 files changed
+838
-81
No files matched your search
+153
-70
@@ -2,16 +2,20 @@ package mcp
|
||||
|
||||
// Agent is the generic OpenAI↔MCP bridge: an OpenAI-compatible chat
|
||||
// completions endpoint whose tools are discovered from — and executed against —
|
||||
// any MCP endpoint. It knows nothing about content, carts, or orders:
|
||||
// one or more MCP endpoints. It knows nothing about content, carts, or orders:
|
||||
//
|
||||
// 1. On every request it calls tools/list on the MCP endpoint and advertises
|
||||
// those tools to the model (merged with any client-supplied tools).
|
||||
// 1. On every request it calls tools/list on each configured MCP endpoint and
|
||||
// advertises those tools to the model (merged with any client-supplied
|
||||
// tools, deduplicated by name — first endpoint wins).
|
||||
// 2. It runs the agentic loop, streaming text deltas to the client as SSE
|
||||
// while intercepting tool calls and dispatching them via tools/call,
|
||||
// appending the configured Bearer token.
|
||||
// 3. Tool calls the MCP server does not know are passed through to the client
|
||||
// while intercepting tool calls and dispatching them via tools/call to the
|
||||
// endpoint that advertised the tool, appending that endpoint's token.
|
||||
// 3. Tool calls no MCP server knows are passed through to the client
|
||||
// (frontend tool calls) so the browser can execute them itself.
|
||||
//
|
||||
// An endpoint that fails tools/list degrades gracefully: it is skipped for that
|
||||
// request (logged), so one unreachable service never takes the assistant down.
|
||||
//
|
||||
// The wire contract for streaming clients: named SSE events "delta" (a JSON
|
||||
// string chunk), <ProgressEvent> (tool activity), "response" (final payload),
|
||||
// then "[DONE]". Non-streaming clients get a normal chat.completion JSON body.
|
||||
@@ -29,6 +33,18 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// MCPEndpoint is one streamable-HTTP MCP server providing tools to the agent.
|
||||
type MCPEndpoint struct {
|
||||
// URL is the JSON-RPC endpoint (e.g. http://127.0.0.1:8941/finder-mcp).
|
||||
URL string
|
||||
// Token is appended as "Authorization: Bearer <Token>" on every request to
|
||||
// this endpoint. Empty means no Authorization header (e.g. loopback).
|
||||
Token string
|
||||
// Label identifies the endpoint in logs and tool-progress events.
|
||||
// Defaults to the URL.
|
||||
Label string
|
||||
}
|
||||
|
||||
// AgentConfig configures an Agent.
|
||||
type AgentConfig struct {
|
||||
// UpstreamURL is the base URL of an OpenAI-compatible API, e.g.
|
||||
@@ -38,11 +54,9 @@ type AgentConfig struct {
|
||||
UpstreamAPIKey string
|
||||
// Model is the default model used when the request omits one.
|
||||
Model string
|
||||
// MCPURL is the streamable-HTTP MCP endpoint providing the tools.
|
||||
MCPURL string
|
||||
// MCPToken is appended as "Authorization: Bearer <MCPToken>" on every MCP
|
||||
// request. Empty means no Authorization header (e.g. a loopback listener).
|
||||
MCPToken string
|
||||
// MCPServers lists the MCP endpoints providing tools. May be empty — the
|
||||
// agent then acts as a plain proxied chat with only LocalTools.
|
||||
MCPServers []MCPEndpoint
|
||||
// SystemPrompt is prepended as a system message unless the conversation
|
||||
// already starts with one. Empty disables injection.
|
||||
SystemPrompt string
|
||||
@@ -52,8 +66,8 @@ type AgentConfig struct {
|
||||
// MaxToolIterations caps the agentic loop (default 10).
|
||||
MaxToolIterations int
|
||||
// LocalTools are executed in-process by the agent instead of being
|
||||
// dispatched over the MCP endpoint. They are advertised to the model
|
||||
// alongside the MCP tools but never reach MCP clients — for capabilities
|
||||
// dispatched over an MCP endpoint. They are advertised to the model
|
||||
// alongside the remote tools but never reach MCP clients — for capabilities
|
||||
// that are agent-only (e.g. a CDP browser tool returning base64
|
||||
// screenshots too bulky and too privileged for the shared tool surface).
|
||||
LocalTools []Tool
|
||||
@@ -61,12 +75,28 @@ type AgentConfig struct {
|
||||
Logger *slog.Logger
|
||||
}
|
||||
|
||||
// mcpRemote is the resolved per-endpoint state used during one request.
|
||||
type mcpRemote struct {
|
||||
endpoint MCPEndpoint
|
||||
url string // trimmed trailing slash
|
||||
token string
|
||||
label string
|
||||
}
|
||||
|
||||
// route is one advertised tool: which endpoint owns it and whether it is a
|
||||
// pure read (from the MCP tool annotations, when the server provides them).
|
||||
type route struct {
|
||||
remote *mcpRemote
|
||||
readOnly bool
|
||||
}
|
||||
|
||||
// Agent is the HTTP handler for the chat completions endpoint. Build it with
|
||||
// NewAgent and mount Handler().
|
||||
type Agent struct {
|
||||
cfg AgentConfig
|
||||
client *http.Client
|
||||
log *slog.Logger
|
||||
cfg AgentConfig
|
||||
client *http.Client
|
||||
log *slog.Logger
|
||||
remotes []mcpRemote
|
||||
}
|
||||
|
||||
// NewAgent builds an Agent.
|
||||
@@ -80,15 +110,34 @@ func NewAgent(cfg AgentConfig) *Agent {
|
||||
if cfg.Logger == nil {
|
||||
cfg.Logger = slog.Default()
|
||||
}
|
||||
remotes := make([]mcpRemote, 0, len(cfg.MCPServers))
|
||||
for _, e := range cfg.MCPServers {
|
||||
if e.URL == "" {
|
||||
continue
|
||||
}
|
||||
label := e.Label
|
||||
if label == "" {
|
||||
label = e.URL
|
||||
}
|
||||
remotes = append(remotes, mcpRemote{
|
||||
endpoint: e,
|
||||
url: strings.TrimSuffix(e.URL, "/"),
|
||||
token: e.Token,
|
||||
label: label,
|
||||
})
|
||||
}
|
||||
return &Agent{
|
||||
cfg: cfg,
|
||||
client: &http.Client{Timeout: 5 * time.Minute},
|
||||
log: cfg.Logger,
|
||||
cfg: cfg,
|
||||
client: &http.Client{Timeout: 5 * time.Minute},
|
||||
log: cfg.Logger,
|
||||
remotes: remotes,
|
||||
}
|
||||
}
|
||||
|
||||
// Enabled reports whether the agent has enough configuration to serve.
|
||||
func (a *Agent) Enabled() bool { return a.cfg.UpstreamURL != "" && a.cfg.UpstreamAPIKey != "" && a.cfg.MCPURL != "" }
|
||||
func (a *Agent) Enabled() bool {
|
||||
return a.cfg.UpstreamURL != "" && a.cfg.UpstreamAPIKey != ""
|
||||
}
|
||||
|
||||
// Handler returns the http.Handler to mount at the chat completions route.
|
||||
func (a *Agent) Handler() http.Handler { return http.HandlerFunc(a.serveHTTP) }
|
||||
@@ -169,13 +218,14 @@ type FrontendToolCall struct {
|
||||
// toolProgress is emitted as a named SSE event for every tool invocation so
|
||||
// streaming clients can show a live activity log.
|
||||
type toolProgress struct {
|
||||
Iter int `json:"iter"`
|
||||
Tool string `json:"tool"`
|
||||
Args string `json:"args,omitempty"`
|
||||
Status string `json:"status"` // "calling" | "done" | "error"
|
||||
Err string `json:"error,omitempty"`
|
||||
Bytes int `json:"bytes,omitempty"`
|
||||
Result any `json:"result,omitempty"`
|
||||
Iter int `json:"iter"`
|
||||
Tool string `json:"tool"`
|
||||
Args string `json:"args,omitempty"`
|
||||
Status string `json:"status"` // "calling" | "done" | "error"
|
||||
Err string `json:"error,omitempty"`
|
||||
Bytes int `json:"bytes,omitempty"`
|
||||
Result any `json:"result,omitempty"`
|
||||
ReadOnly bool `json:"readOnly,omitempty"`
|
||||
}
|
||||
|
||||
// ── HTTP handler ───────────────────────────────────────────────────────────────
|
||||
@@ -200,12 +250,10 @@ func (a *Agent) serveHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
req.Model = a.cfg.Model
|
||||
}
|
||||
|
||||
mcpTools, err := a.listTools(r.Context())
|
||||
if err != nil {
|
||||
a.log.Error("agent: tools/list failed", "err", err)
|
||||
http.Error(w, "mcp tools/list: "+err.Error(), http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
// Discover tools from every endpoint. A failing endpoint is skipped for
|
||||
// this request (logged) — one unreachable service never takes the
|
||||
// assistant down. routes maps tool name → the endpoint that advertised it.
|
||||
mcpTools, routes := a.listTools(r.Context())
|
||||
req = mergeTools(req, mcpTools, localToolDefs(a.cfg.LocalTools))
|
||||
if a.cfg.SystemPrompt != "" && (len(req.Messages) == 0 || req.Messages[0].Role != "system") {
|
||||
req.Messages = append([]Message{{Role: "system", Content: a.cfg.SystemPrompt}}, req.Messages...)
|
||||
@@ -242,7 +290,7 @@ func (a *Agent) serveHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, msg, status)
|
||||
}
|
||||
|
||||
content, finishReason, frontendCalls, ok := a.loop(r.Context(), w, flush, sendErr, &req)
|
||||
content, finishReason, frontendCalls, ok := a.loop(r.Context(), w, flush, sendErr, &req, routes)
|
||||
if !ok {
|
||||
return // sendErr already wrote the response
|
||||
}
|
||||
@@ -286,6 +334,7 @@ func (a *Agent) loop(
|
||||
flush func(),
|
||||
sendErr func(status int, msg string),
|
||||
req *ChatRequest,
|
||||
routes map[string]route,
|
||||
) (content, finishReason string, frontendCalls []FrontendToolCall, ok bool) {
|
||||
parallel := true
|
||||
req.ParallelToolCalls = ¶llel
|
||||
@@ -341,9 +390,16 @@ func (a *Agent) loop(
|
||||
local := localToolIndex(a.cfg.LocalTools)
|
||||
var remaining []ToolCall
|
||||
for _, tc := range toolCalls {
|
||||
a.emitProgress(w, flush, toolProgress{
|
||||
progress := toolProgress{
|
||||
Iter: iter, Tool: tc.Function.Name, Args: tc.Function.Arguments, Status: "calling",
|
||||
})
|
||||
}
|
||||
if lt, isLocal := local[tc.Function.Name]; isLocal && lt.Annotations != nil {
|
||||
progress.ReadOnly = lt.Annotations.ReadOnlyHint
|
||||
}
|
||||
if rt, isRemote := routes[tc.Function.Name]; isRemote && rt.remote != nil {
|
||||
progress.ReadOnly = rt.readOnly
|
||||
}
|
||||
a.emitProgress(w, flush, progress)
|
||||
|
||||
var (
|
||||
resultContent string
|
||||
@@ -361,13 +417,16 @@ func (a *Agent) loop(
|
||||
b, _ := json.Marshal(out)
|
||||
resultContent = string(b)
|
||||
}
|
||||
} else {
|
||||
out, found, callToolErr := a.callTool(ctx, tc.Function.Name, tc.Function.Arguments)
|
||||
} else if rt, ok := routes[tc.Function.Name]; ok {
|
||||
out, found, callToolErr := a.callTool(ctx, rt.remote, tc.Function.Name, tc.Function.Arguments)
|
||||
if !found {
|
||||
remaining = append(remaining, tc)
|
||||
continue
|
||||
}
|
||||
resultContent, isError, callErr = out.text, out.isError, callToolErr
|
||||
} else {
|
||||
remaining = append(remaining, tc)
|
||||
continue
|
||||
}
|
||||
|
||||
status, summary := "done", any(nil)
|
||||
@@ -591,9 +650,10 @@ func localToolIndex(tools []Tool) map[string]Tool {
|
||||
|
||||
type mcpToolList struct {
|
||||
Tools []struct {
|
||||
Name string `json:"name"`
|
||||
Description string `json:"description"`
|
||||
InputSchema json.RawMessage `json:"inputSchema"`
|
||||
Name string `json:"name"`
|
||||
Description string `json:"description"`
|
||||
InputSchema json.RawMessage `json:"inputSchema"`
|
||||
Annotations *ToolAnnotations `json:"annotations"`
|
||||
} `json:"tools"`
|
||||
}
|
||||
|
||||
@@ -611,30 +671,50 @@ type toolOutput struct {
|
||||
isError bool // MCP isError flag (a tool-level failure the model can read)
|
||||
}
|
||||
|
||||
// listTools fetches the MCP server's advertised tools as OpenAI definitions.
|
||||
func (a *Agent) listTools(ctx context.Context) ([]ToolDef, error) {
|
||||
var out mcpToolList
|
||||
if err := a.mcpRequest(ctx, "tools/list", nil, &out); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defs := make([]ToolDef, 0, len(out.Tools))
|
||||
for _, t := range out.Tools {
|
||||
schema := t.InputSchema
|
||||
if len(schema) == 0 {
|
||||
schema = Object(nil, nil)
|
||||
// listTools fetches tools/list from every configured endpoint. It returns the
|
||||
// merged OpenAI definitions (first endpoint wins on name collisions) plus the
|
||||
// routes map tool name → endpoint used for dispatch.
|
||||
func (a *Agent) listTools(ctx context.Context) ([]ToolDef, map[string]route) {
|
||||
defs := make([]ToolDef, 0)
|
||||
routes := make(map[string]route)
|
||||
seen := make(map[string]bool)
|
||||
for i := range a.remotes {
|
||||
remote := &a.remotes[i]
|
||||
var out mcpToolList
|
||||
if err := remote.request(ctx, a.client, "tools/list", nil, &out); err != nil {
|
||||
a.log.Warn("agent: tools/list failed — skipping endpoint for this request",
|
||||
"endpoint", remote.label, "err", err)
|
||||
continue
|
||||
}
|
||||
added := 0
|
||||
for _, t := range out.Tools {
|
||||
if t.Name == "" || seen[t.Name] {
|
||||
continue // first endpoint wins on collisions
|
||||
}
|
||||
schema := t.InputSchema
|
||||
if len(schema) == 0 {
|
||||
schema = Object(nil, nil)
|
||||
}
|
||||
defs = append(defs, ToolDef{Type: "function", Function: FuncDef{
|
||||
Name: t.Name,
|
||||
Description: t.Description,
|
||||
Parameters: schema,
|
||||
}})
|
||||
routes[t.Name] = route{remote: remote, readOnly: t.Annotations != nil && t.Annotations.ReadOnlyHint}
|
||||
seen[t.Name] = true
|
||||
added++
|
||||
}
|
||||
if added > 0 {
|
||||
a.log.Info("agent: mcp endpoint merged", "endpoint", remote.label, "tools", added)
|
||||
}
|
||||
defs = append(defs, ToolDef{Type: "function", Function: FuncDef{
|
||||
Name: t.Name,
|
||||
Description: t.Description,
|
||||
Parameters: schema,
|
||||
}})
|
||||
}
|
||||
return defs, nil
|
||||
return defs, routes
|
||||
}
|
||||
|
||||
// callTool dispatches one tools/call. found=false means the MCP server does not
|
||||
// know the tool (the caller passes it through to the frontend).
|
||||
func (a *Agent) callTool(ctx context.Context, name, argsJSON string) (out toolOutput, found bool, err error) {
|
||||
// callTool dispatches one tools/call to the endpoint that advertised the tool.
|
||||
// found=false means the route vanished (the caller passes the call through to
|
||||
// the frontend).
|
||||
func (a *Agent) callTool(ctx context.Context, remote *mcpRemote, name, argsJSON string) (out toolOutput, found bool, err error) {
|
||||
params := map[string]any{"name": name}
|
||||
if strings.TrimSpace(argsJSON) != "" {
|
||||
var args any
|
||||
@@ -643,11 +723,11 @@ func (a *Agent) callTool(ctx context.Context, name, argsJSON string) (out toolOu
|
||||
}
|
||||
}
|
||||
var res mcpCallResult
|
||||
if err := a.mcpRequest(ctx, "tools/call", params, &res); err != nil {
|
||||
if err := remote.request(ctx, a.client, "tools/call", params, &res); err != nil {
|
||||
if strings.Contains(err.Error(), "unknown tool") {
|
||||
return toolOutput{}, false, nil
|
||||
}
|
||||
return toolOutput{}, true, err
|
||||
return toolOutput{}, true, fmt.Errorf("%s: %w", remote.label, err)
|
||||
}
|
||||
|
||||
var texts strings.Builder
|
||||
@@ -686,9 +766,9 @@ func previewOf(s string) any {
|
||||
}
|
||||
}
|
||||
|
||||
// mcpRequest posts one JSON-RPC request to the MCP endpoint, appending the
|
||||
// configured Bearer token, and decodes the result into out.
|
||||
func (a *Agent) mcpRequest(ctx context.Context, method string, params, out any) error {
|
||||
// request posts one JSON-RPC request to this endpoint, appending its Bearer
|
||||
// token, and decodes the result into out.
|
||||
func (r *mcpRemote) request(ctx context.Context, client *http.Client, method string, params, out any) error {
|
||||
body, err := json.Marshal(map[string]any{
|
||||
"jsonrpc": "2.0",
|
||||
"id": 1,
|
||||
@@ -698,16 +778,19 @@ func (a *Agent) mcpRequest(ctx context.Context, method string, params, out any)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, a.cfg.MCPURL, bytes.NewReader(body))
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, r.url, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Accept", "application/json")
|
||||
if a.cfg.MCPToken != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+a.cfg.MCPToken)
|
||||
if r.token != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+r.token)
|
||||
}
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if client == nil {
|
||||
client = http.DefaultClient
|
||||
}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("mcp %s: %w", method, err)
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user