NewRequestBridge creates a new *[RequestBridge] and registers the HTTP routes defined by the given providers. Any routes which are requested but not registered will be reverse-proxied to the upstream service. A [intercept.Recorder] is also required to record prompt, tool, and token use. mcpProxy w
(ctx context.Context, providers []provider.Provider, rec recorder.Recorder, mcpProxy mcp.ServerProxier, logger slog.Logger, m *metrics.Metrics, tracer trace.Tracer)
| 89 | // Circuit breaker configuration is obtained from each provider's CircuitBreakerConfig() method. |
| 90 | // Providers returning nil will not have circuit breaker protection. |
| 91 | func NewRequestBridge(ctx context.Context, providers []provider.Provider, rec recorder.Recorder, mcpProxy mcp.ServerProxier, logger slog.Logger, m *metrics.Metrics, tracer trace.Tracer) (*RequestBridge, error) { |
| 92 | if err := validateProviders(providers); err != nil { |
| 93 | return nil, err |
| 94 | } |
| 95 | |
| 96 | mux := http.NewServeMux() |
| 97 | |
| 98 | for _, prov := range providers { |
| 99 | // Create per-provider circuit breaker if configured |
| 100 | cfg := prov.CircuitBreakerConfig() |
| 101 | providerName := prov.Name() |
| 102 | onChange := func(endpoint, model string, from, to gobreaker.State) { |
| 103 | logger.Info(context.Background(), "circuit breaker state change", |
| 104 | slog.F("provider", providerName), |
| 105 | slog.F("endpoint", endpoint), |
| 106 | slog.F("model", model), |
| 107 | slog.F("from", from.String()), |
| 108 | slog.F("to", to.String()), |
| 109 | ) |
| 110 | if m != nil { |
| 111 | m.CircuitBreakerState.WithLabelValues(providerName, endpoint, model).Set(circuitbreaker.StateToGaugeValue(to)) |
| 112 | if to == gobreaker.StateOpen { |
| 113 | m.CircuitBreakerTrips.WithLabelValues(providerName, endpoint, model).Inc() |
| 114 | } |
| 115 | } |
| 116 | } |
| 117 | cbs := circuitbreaker.NewProviderCircuitBreakers(providerName, cfg, onChange, m) |
| 118 | |
| 119 | // Add the known provider-specific routes which are bridged (i.e. intercepted and augmented). |
| 120 | for _, path := range prov.BridgedRoutes() { |
| 121 | handler := newInterceptionProcessor(prov, cbs, rec, mcpProxy, logger, m, tracer) |
| 122 | route, err := url.JoinPath(prov.RoutePrefix(), path) |
| 123 | if err != nil { |
| 124 | logger.Error(ctx, "failed to join path", |
| 125 | slog.Error(err), |
| 126 | slog.F("provider", providerName), |
| 127 | slog.F("prefix", prov.RoutePrefix()), |
| 128 | slog.F("path", path), |
| 129 | ) |
| 130 | return nil, xerrors.Errorf("failed to configure provider '%v': failed to join bridged path: %w", providerName, err) |
| 131 | } |
| 132 | mux.Handle(route, handler) |
| 133 | } |
| 134 | |
| 135 | // Any requests which passthrough to this will be reverse-proxied to the upstream. |
| 136 | // |
| 137 | // We have to whitelist the known-safe routes because an API key with elevated privileges (i.e. admin) might be |
| 138 | // configured, so we should just reverse-proxy known-safe routes. |
| 139 | ftr := newPassthroughRouter(prov, logger.Named(fmt.Sprintf("passthrough.%s", prov.Name())), m, tracer) |
| 140 | for _, path := range prov.PassthroughRoutes() { |
| 141 | route, err := url.JoinPath(prov.RoutePrefix(), path) |
| 142 | if err != nil { |
| 143 | logger.Error(ctx, "failed to join path", |
| 144 | slog.Error(err), |
| 145 | slog.F("provider", providerName), |
| 146 | slog.F("prefix", prov.RoutePrefix()), |
| 147 | slog.F("path", path), |
| 148 | ) |