newInterceptionProcessor returns an [http.HandlerFunc] which is capable of creating a new interceptor and processing a given request using [Provider] p, recording all usage events using [Recorder] rec. If cbs is non-nil, circuit breaker protection is applied per endpoint/model tuple.
(p provider.Provider, cbs *circuitbreaker.ProviderCircuitBreakers, rec recorder.Recorder, mcpProxy mcp.ServerProxier, logger slog.Logger, m *metrics.Metrics, tracer trace.Tracer)
| 174 | // using [Provider] p, recording all usage events using [Recorder] rec. |
| 175 | // If cbs is non-nil, circuit breaker protection is applied per endpoint/model tuple. |
| 176 | func newInterceptionProcessor(p provider.Provider, cbs *circuitbreaker.ProviderCircuitBreakers, rec recorder.Recorder, mcpProxy mcp.ServerProxier, logger slog.Logger, m *metrics.Metrics, tracer trace.Tracer) http.HandlerFunc { |
| 177 | return func(w http.ResponseWriter, r *http.Request) { |
| 178 | ctx, span := tracer.Start(r.Context(), "Intercept") |
| 179 | defer span.End() |
| 180 | |
| 181 | // We execute this before CreateInterceptor since the interceptors |
| 182 | // read the request body and don't reset them. |
| 183 | client := GuessClient(r) |
| 184 | sessionID := GuessSessionID(client, r) |
| 185 | |
| 186 | interceptor, err := p.CreateInterceptor(w, r.WithContext(ctx), tracer) |
| 187 | if err != nil { |
| 188 | span.SetStatus(codes.Error, fmt.Sprintf("failed to create interceptor: %v", err)) |
| 189 | logger.Warn(ctx, "failed to create interceptor", slog.Error(err), slog.F("path", r.URL.Path)) |
| 190 | http.Error(w, fmt.Sprintf("failed to create %q interceptor", r.URL.Path), http.StatusInternalServerError) |
| 191 | return |
| 192 | } |
| 193 | |
| 194 | if m != nil { |
| 195 | start := time.Now() |
| 196 | defer func() { |
| 197 | m.InterceptionDuration.WithLabelValues(p.Name(), interceptor.Model()).Observe(time.Since(start).Seconds()) |
| 198 | }() |
| 199 | } |
| 200 | |
| 201 | actor := aibcontext.ActorFromContext(ctx) |
| 202 | if actor == nil { |
| 203 | logger.Warn(ctx, "no actor found in context") |
| 204 | http.Error(w, "no actor found", http.StatusBadRequest) |
| 205 | return |
| 206 | } |
| 207 | |
| 208 | traceAttrs := interceptor.TraceAttributes(r) |
| 209 | span.SetAttributes(traceAttrs...) |
| 210 | ctx = tracing.WithInterceptionAttributesInContext(ctx, traceAttrs) |
| 211 | r = r.WithContext(ctx) |
| 212 | |
| 213 | // Record usage in the background to not block request flow. |
| 214 | asyncRecorder := recorder.NewAsyncRecorder(logger, rec, recordingTimeout) |
| 215 | asyncRecorder.WithMetrics(m) |
| 216 | asyncRecorder.WithProvider(p.Name()) |
| 217 | asyncRecorder.WithModel(interceptor.Model()) |
| 218 | asyncRecorder.WithInitiatorID(actor.ID) |
| 219 | asyncRecorder.WithClient(string(client)) |
| 220 | interceptor.Setup(logger, asyncRecorder, mcpProxy) |
| 221 | |
| 222 | cred := interceptor.Credential() |
| 223 | if err := rec.RecordInterception(ctx, &recorder.InterceptionRecord{ |
| 224 | ID: interceptor.ID().String(), |
| 225 | InitiatorID: actor.ID, |
| 226 | Metadata: actor.Metadata, |
| 227 | Model: interceptor.Model(), |
| 228 | Provider: p.Type(), |
| 229 | ProviderName: p.Name(), |
| 230 | UserAgent: r.UserAgent(), |
| 231 | Client: string(client), |
| 232 | ClientSessionID: sessionID, |
| 233 | CorrelatingToolCallID: interceptor.CorrelatingToolCallID(), |
no test coverage detected