agent endpoints: serve webtransport from the node's server set

StartWebTransport built the node's only HTTP/3 listener from inside AgentService:
it minted the TLS config, bound the UDP socket, assembled an h3 mux and returned
a stop closure LivekitServer held in agentWTStop. Every other listener is
constructed in NewLivekitServer and delegates to a service, and agents are the
first consumer of HTTP/3 rather than the last.

webtransport.go now carries the transport with no agent references:
WebTransportTLS, NewWebTransportServer, NewWebTransportHandler,
ListenWebTransport, and WithWebTransportServer/GetWebTransportServer, which put
the serving server in the request context so a route that upgrades is a
conventional http.Handler instead of a closure over the server. The listener is
built in NewLivekitServer beside the prometheus and debug servers, bound in Start
with the other listeners, and stopped once doneChan unblocks, so no stop closure
is held on LivekitServer. /agent is a route on its mux, as it is on the API mux.

UpgradeWebTransport unwraps the ResponseWriter before upgrading. webtransport-go
type-asserts it to http3.Settingser and http3.HTTPStreamer without checking and
negroni wraps the writer on every chain, so the h3 listener could not carry a
middleware chain at all. It now runs the same recovery, api-key auth and path
normalization as the TCP chain.

AgentService splits into AgentWSService and AgentWTService over the shared
AgentHandler, and the endpoint front becomes AgentEndpointService. wire builds
all four and threads one endpoint.Registry into the handler and the front.
NewAgentHandler takes the config and builds its own ServerInfo, so endpointsConfig
and singleAPIKey are set at construction and the psrpc server is registered
against a fully formed handler; it previously bound one whose embedded
*AgentHandler was still nil.

The listener's port and certificate move to a top-level webtransport config
block. The listener no longer reads agents.endpoints.disabled, which continues to
refuse endpoint registrations. ListenWebTransport binds one socket per bind
address, where the h3 listener ignored bind_addresses that every TCP listener
honoured.

NewWorkerRegisterer takes a variadic WorkerRegisterHandler, each handler reading
the request and filling its part of the response and the registration, replacing
the single EndpointSettingsFunc threaded through HandleRegister.
EndpointRegisterHandler is the endpoint declaration's turn, and
endpoint.NegotiateSettings holds the protocol negotiation the OSS and cloud
handshakes each carried a copy of.

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

pkg/agent/config.go

@@ -17,15 +17,6 @@ type EndpointsConfig struct {
 	// rejects registrations that declare endpoints.
 	Disabled bool `yaml:"disabled,omitempty"`
 
-	// WebTransportPort is the UDP port for the worker WebTransport listener that
-	// carries the control stream and HTTP exchanges. 0 disables the listener, so
-	// no worker can serve endpoints (the front then only relays, in cloud).
-	WebTransportPort uint32 `yaml:"webtransport_port,omitempty"`
-	// TLSCertFile / TLSKeyFile is the certificate the WebTransport (QUIC) listener
-	// presents. QUIC has no plaintext mode, so a listener needs one; in dev mode a
-	// self-signed cert is generated when these are unset.
-	TLSCertFile string `yaml:"tls_cert_file,omitempty"`
-	TLSKeyFile  string `yaml:"tls_key_file,omitempty"`
 	// MaxStreams is the soft per-session concurrent-stream cap used for capacity
 	// weighting; 0 takes the endpoint package default.
 	MaxStreams uint32 `yaml:"max_streams,omitempty"`

pkg/agent/endpoint/manifest.go

@@ -15,12 +15,15 @@
 package endpoint
 
 import (
+	"errors"
 	"fmt"
 	"net/http"
 	"slices"
 	"strings"
 
 	"github.com/livekit/protocol/livekit"
+
+	"github.com/livekit/livekit-server/pkg/agent/endpoint/wire"
 )
 
 const MaxManifestRoutes = 256
@@ -64,6 +67,30 @@ const (
 )
 
 // ParseManifest validates a registration's endpoint list.
+// NegotiateSettings validates a registration's endpoint manifest and negotiates
+// the data-plane protocol version. The data plane is a WebTransport session (no
+// attach token, no fixed connection pool), so the version is all there is to
+// agree on.
+func NegotiateSettings(req *livekit.RegisterWorkerRequest) (*livekit.AgentHttp_AgentEndpointSettings, error) {
+	if _, err := ParseManifest(req.GetEndpoints()); err != nil {
+		return nil, err
+	}
+	if req.GetInstanceId() == "" {
+		return nil, errors.New("registrations with endpoints require an instance_id")
+	}
+	if req.GetEndpointProtocol() == 0 {
+		return nil, errors.New("worker declared endpoints but no endpoint protocol; upgrade the agent SDK to one that speaks the endpoint data plane")
+	}
+	// the worker frames to the version returned here, so it must be the
+	// negotiated one and not the server's own constant
+	negotiated := min(req.GetEndpointProtocol(), wire.CurrentProtocol)
+	if negotiated < wire.MinProtocol {
+		return nil, fmt.Errorf("unsupported agent endpoint protocol %d (this server serves %d..%d)",
+			req.GetEndpointProtocol(), wire.MinProtocol, wire.CurrentProtocol)
+	}
+	return &livekit.AgentHttp_AgentEndpointSettings{Protocol: negotiated}, nil
+}
+
 func ParseManifest(endpoints []*livekit.AgentHttp_AgentEndpoint) (*Manifest, error) {
 	if len(endpoints) > MaxManifestRoutes {
 		return nil, fmt.Errorf("manifest exceeds %d routes", MaxManifestRoutes)

pkg/agent/testutils/server.go

@@ -15,6 +15,7 @@ import (
 	"github.com/gammazero/deque"
 
 	"github.com/livekit/livekit-server/pkg/agent"
+	"github.com/livekit/livekit-server/pkg/agent/endpoint"
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/routing"
 	"github.com/livekit/livekit-server/pkg/service"
@@ -40,7 +41,7 @@ type TestServer struct {
 
 func NewTestServer(bus psrpc.MessageBus) *TestServer {
 	localNode, _ := routing.NewLocalNode(nil)
-	return NewTestServerWithService(must.Get(service.NewAgentService(
+	return NewTestServerWithService(must.Get(service.NewAgentHandler(
 		&config.Config{
 			Region: "test",
 			Agents: agent.Config{
@@ -50,6 +51,7 @@ func NewTestServer(bus psrpc.MessageBus) *TestServer {
 		localNode,
 		bus,
 		auth.NewSimpleKeyProvider("test", "verysecretsecret"),
+		endpoint.NewRegistry(),
 	)))
 }
 

pkg/agent/worker.go

@@ -179,29 +179,52 @@ func MakeWorkerRegistration() WorkerRegistration {
 
 var _ WorkerSignalHandler = (*WorkerRegisterer)(nil)
 
-// EndpointSettingsFunc validates a registration's endpoint manifest and returns the
-// negotiated data-plane settings (including the attach token). Returning an error
-// fails the registration. It runs only when the registration declares endpoints AND
-// a data-plane protocol version; a nil func rejects such registrations.
-type EndpointSettingsFunc func(req *livekit.RegisterWorkerRequest) (*livekit.AgentHttp_AgentEndpointSettings, error)
+// WorkerRegisterHandler takes a turn during a worker's registration: it reads the
+// request, validates the part it owns, and fills in its part of the response and
+// the registration. Returning an error fails the registration.
+type WorkerRegisterHandler func(req *livekit.RegisterWorkerRequest, res *livekit.RegisterWorkerResponse, reg *WorkerRegistration) error
+
+// EndpointRegisterHandler owns a worker's endpoint declaration: it validates the
+// declaration and negotiates the data-plane settings. A registration declaring no
+// endpoints passes through untouched.
+func EndpointRegisterHandler(req *livekit.RegisterWorkerRequest, res *livekit.RegisterWorkerResponse, reg *WorkerRegistration) error {
+	if len(req.GetEndpoints()) == 0 {
+		return nil
+	}
+	// endpoints are addressed at /agents/{agent_name}/{deployment}/..., with the
+	// name percent-encoded into that segment
+	if req.GetAgentName() == "" {
+		return errors.New("agent HTTP endpoints require an agent name")
+	}
+	if endpoint.IsReservedAgentName(req.GetAgentName()) {
+		return fmt.Errorf("agent name %q is reserved and cannot serve HTTP endpoints", req.GetAgentName())
+	}
+	settings, err := endpoint.NegotiateSettings(req)
+	if err != nil {
+		return err
+	}
+	res.EndpointSettings = settings
+	reg.EndpointSettings = settings
+	return nil
+}
 
 type WorkerRegisterer struct {
 	WorkerPingHandler
-	serverInfo       *livekit.ServerInfo
-	deadline         time.Time
-	endpointSettings EndpointSettingsFunc
+	serverInfo *livekit.ServerInfo
+	deadline   time.Time
+	handlers   []WorkerRegisterHandler
 
 	registration WorkerRegistration
 	registered   bool
 }
 
-func NewWorkerRegisterer(conn SignalConn, serverInfo *livekit.ServerInfo, base WorkerRegistration, endpointSettings EndpointSettingsFunc) *WorkerRegisterer {
+func NewWorkerRegisterer(conn SignalConn, serverInfo *livekit.ServerInfo, base WorkerRegistration, handlers ...WorkerRegisterHandler) *WorkerRegisterer {
 	return &WorkerRegisterer{
 		WorkerPingHandler: WorkerPingHandler{conn: conn},
 		serverInfo:        serverInfo,
 		registration:      base,
 		deadline:          time.Now().Add(RegisterTimeout),
-		endpointSettings:  endpointSettings,
+		handlers:          handlers,
 	}
 }
 
@@ -245,34 +268,19 @@ func (h *WorkerRegisterer) HandleRegister(req *livekit.RegisterWorkerRequest) er
 	h.registration.Endpoints = req.GetEndpoints()
 	h.registration.InstanceID = req.GetInstanceId()
 
-	if len(req.GetEndpoints()) > 0 {
-		if h.endpointSettings == nil {
-			return errors.New("agent HTTP endpoints are not supported by this server")
-		}
-		// endpoints are addressed at /agents/{agent_name}/{deployment}/..., with
-		// the name percent-encoded into that segment
-		if req.GetAgentName() == "" {
-			return errors.New("agent HTTP endpoints require an agent name")
-		}
-		if endpoint.IsReservedAgentName(req.GetAgentName()) {
-			return fmt.Errorf("agent name %q is reserved and cannot serve HTTP endpoints", req.GetAgentName())
-		}
-		settings, err := h.endpointSettings(req)
-		if err != nil {
+	res := &livekit.RegisterWorkerResponse{
+		WorkerId:   h.registration.ID,
+		ServerInfo: h.serverInfo,
+	}
+	for _, handle := range h.handlers {
+		if err := handle(req, res, &h.registration); err != nil {
 			return err
 		}
-		h.registration.EndpointSettings = settings
 	}
 	h.registered = true
 
 	_, err := h.conn.WriteServerMessage(&livekit.ServerMessage{
-		Message: &livekit.ServerMessage_Register{
-			Register: &livekit.RegisterWorkerResponse{
-				WorkerId:         h.registration.ID,
-				ServerInfo:       h.serverInfo,
-				EndpointSettings: h.registration.EndpointSettings,
-			},
-		},
+		Message: &livekit.ServerMessage_Register{Register: res},
 	})
 	return err
 }

pkg/agent/worker_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/livekit/livekit-server/pkg/agent"
+	"github.com/livekit/livekit-server/pkg/agent/endpoint/wire"
 	"github.com/livekit/protocol/livekit"
 )
 
@@ -47,14 +48,14 @@ func TestHandleRegisterEndpointAgentNames(t *testing.T) {
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
 			h := agent.NewWorkerRegisterer(nopSignalConn{}, &livekit.ServerInfo{}, agent.WorkerRegistration{},
-				func(*livekit.RegisterWorkerRequest) (*livekit.AgentHttp_AgentEndpointSettings, error) {
-					return &livekit.AgentHttp_AgentEndpointSettings{}, nil
-				})
+				agent.EndpointRegisterHandler)
 
 			err := h.HandleRegister(&livekit.RegisterWorkerRequest{
-				Type:      livekit.JobType_JT_ROOM,
-				AgentName: c.agentName,
-				Endpoints: []*livekit.AgentHttp_AgentEndpoint{{Path: "/hook", Methods: []string{"GET"}}},
+				Type:             livekit.JobType_JT_ROOM,
+				AgentName:        c.agentName,
+				Endpoints:        []*livekit.AgentHttp_AgentEndpoint{{Path: "/hook", Methods: []string{"GET"}}},
+				InstanceId:       "AEI_test",
+				EndpointProtocol: wire.CurrentProtocol,
 			})
 			if c.ok {
 				require.NoError(t, err)
@@ -68,7 +69,7 @@ func TestHandleRegisterEndpointAgentNames(t *testing.T) {
 // the reserved names constrain only workers that declare endpoints.
 func TestHandleRegisterWithoutEndpointsIgnoresReservedNames(t *testing.T) {
 	for _, name := range []string{"", "_", ".", ".."} {
-		h := agent.NewWorkerRegisterer(nopSignalConn{}, &livekit.ServerInfo{}, agent.WorkerRegistration{}, nil)
+		h := agent.NewWorkerRegisterer(nopSignalConn{}, &livekit.ServerInfo{}, agent.WorkerRegistration{}, agent.EndpointRegisterHandler)
 		require.NoError(t, h.HandleRegister(&livekit.RegisterWorkerRequest{
 			Type:      livekit.JobType_JT_ROOM,
 			AgentName: name,

pkg/config/config.go

@@ -82,6 +82,7 @@ type Config struct {
 	PrometheusPort uint32                   `yaml:"prometheus_port,omitempty"`
 	Prometheus     PrometheusConfig         `yaml:"prometheus,omitempty"`
 	DebugHandler   DebugHandlerConfig       `yaml:"debug_handler,omitempty"`
+	WebTransport   WebTransportConfig       `yaml:"webtransport,omitempty"`
 	RTC            RTCConfig                `yaml:"rtc,omitempty"`
 	Redis          redisLiveKit.RedisConfig `yaml:"redis,omitempty"`
 	Audio          sfu.AudioConfig          `yaml:"audio,omitempty"`
@@ -117,6 +118,15 @@ type Config struct {
 	API APIConfig `yaml:"api,omitempty"`
 }
 
+// WebTransportConfig is the node's HTTP/3 listener. QUIC has no plaintext mode,
+// so a certificate is required; development mode generates a self-signed one.
+type WebTransportConfig struct {
+	// Port is the UDP port to listen on. 0 starts no listener.
+	Port        uint32 `yaml:"port,omitempty"`
+	TLSCertFile string `yaml:"tls_cert_file,omitempty"`
+	TLSKeyFile  string `yaml:"tls_key_file,omitempty"`
+}
+
 type RTCConfig struct {
 	rtcconfig.RTCConfig `yaml:",inline"`
 

pkg/service/agentendpoint.go

@@ -0,0 +1,50 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package service
+
+import (
+	"net/http"
+
+	"github.com/livekit/livekit-server/pkg/agent/endpoint"
+)
+
+// AgentEndpointService is the /agents/{agent_name}/{deployment}/{path...} front
+// backed by this node's attached workers. The api key comes from validated grants
+// when a token is present; a non-public route additionally requires an
+// agent-endpoint grant scoped to this agent and deployment.
+type AgentEndpointService struct {
+	*endpoint.Front
+}
+
+func NewAgentEndpointService(h *AgentHandler, registry *endpoint.Registry) *AgentEndpointService {
+	return &AgentEndpointService{
+		Front: endpoint.NewFront(endpoint.FrontParams{
+			Registry: registry,
+			ResolveAccess: func(r *http.Request, agentName, deployment string) endpoint.Access {
+				if claims := GetGrants(r.Context()); claims != nil {
+					level := endpoint.AccessCredentialed
+					if claims.AgentEndpoint.Allows(agentName, deployment) {
+						level = endpoint.AccessGranted
+					}
+					return endpoint.Access{APIKey: GetAPIKey(r.Context()), Level: level}
+				}
+				// unauthenticated: one configured key makes the api key unambiguous
+				return endpoint.Access{APIKey: h.singleAPIKey}
+			},
+			Logger:            h.logger,
+			SingleKeyFallback: true,
+		}),
+	}
+}

pkg/service/agentendpoint_protocol_test.go

@@ -35,29 +35,46 @@ func TestEndpointSettingsNegotiation(t *testing.T) {
 		}
 	}
 	h := &AgentHandler{}
+	negotiate := func(h *AgentHandler, r *livekit.RegisterWorkerRequest) (*livekit.AgentHttp_AgentEndpointSettings, error) {
+		res := &livekit.RegisterWorkerResponse{}
+		reg := &agent.WorkerRegistration{}
+		if err := h.endpointRegisterHandler(r, res, reg); err != nil {
+			return nil, err
+		}
+		require.Equal(t, res.GetEndpointSettings(), reg.EndpointSettings,
+			"the registration must carry what the worker was told")
+		return res.GetEndpointSettings(), nil
+	}
 
 	// a worker predating the data plane declares endpoints with protocol 0: it
 	// cannot frame at all, and the error must name that rather than read as a
 	// version out of range
-	_, err := h.endpointSettings(req(0))
+	_, err := negotiate(h, req(0))
 	require.ErrorContains(t, err, "no endpoint protocol")
 
-	settings, err := h.endpointSettings(req(wire.CurrentProtocol))
+	settings, err := negotiate(h, req(wire.CurrentProtocol))
 	require.NoError(t, err)
 	require.Equal(t, wire.CurrentProtocol, settings.GetProtocol())
 
 	// a worker speaking a newer version is negotiated down to what this server
 	// serves; the worker frames to the version returned here
-	settings, err = h.endpointSettings(req(wire.CurrentProtocol + 42))
+	settings, err = negotiate(h, req(wire.CurrentProtocol+42))
 	require.NoError(t, err)
 	require.Equal(t, wire.CurrentProtocol, settings.GetProtocol())
 
 	noInstance := req(wire.CurrentProtocol)
 	noInstance.InstanceId = ""
-	_, err = h.endpointSettings(noInstance)
+	_, err = negotiate(h, noInstance)
 	require.ErrorContains(t, err, "instance_id")
 
 	disabled := &AgentHandler{endpointsConfig: agent.EndpointsConfig{Disabled: true}}
-	_, err = disabled.endpointSettings(req(wire.CurrentProtocol))
+	_, err = negotiate(disabled, req(wire.CurrentProtocol))
 	require.ErrorContains(t, err, "disabled")
+
+	// a registration declaring no endpoints is left alone, even with endpoints off
+	bare := req(wire.CurrentProtocol)
+	bare.Endpoints = nil
+	settings, err = negotiate(disabled, bare)
+	require.NoError(t, err)
+	require.Nil(t, settings)
 }

pkg/service/agentendpoint_test.go

@@ -41,6 +41,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/livekit/livekit-server/pkg/agent"
+	"github.com/livekit/livekit-server/pkg/agent/endpoint"
 	"github.com/livekit/livekit-server/pkg/agent/endpoint/conformance"
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/routing"
@@ -59,10 +60,10 @@ const (
 const testMaxAPIBodySize = 64 << 10
 
 type endpointStack struct {
-	t     *testing.T
-	ts    *httptest.Server
-	svc   *service.AgentService
-	wtURL string // https://host:port/agent (WebTransport control+data)
+	t       *testing.T
+	ts      *httptest.Server
+	handler *service.AgentHandler
+	wtURL   string // https://host:port/agent (WebTransport control+data)
 }
 
 // selfSignedTLS mints an in-memory cert for 127.0.0.1 with the h3 ALPN, for the
@@ -97,7 +98,8 @@ func newEndpointStack(t *testing.T, endpointsCfg agent.EndpointsConfig) *endpoin
 	}
 	conf.Limit.MaxAPIRequestBodySize = testMaxAPIBodySize
 
-	svc, err := service.NewAgentService(conf, localNode, psrpc.NewLocalMessageBus(), keyProvider)
+	registry := endpoint.NewRegistry()
+	h, err := service.NewAgentHandler(conf, localNode, psrpc.NewLocalMessageBus(), keyProvider, registry)
 	require.NoError(t, err)
 
 	// the production handler, so these tests run on the node's real middleware chain
@@ -110,21 +112,23 @@ func newEndpointStack(t *testing.T, endpointsCfg agent.EndpointsConfig) *endpoin
 
 	var agentFront http.Handler
 	if !endpointsCfg.Disabled {
-		agentFront = svc.EndpointFront()
+		agentFront = service.NewAgentEndpointService(h, registry)
 	}
 	ts := httptest.NewServer(service.NewHTTPHandler(conf, keyProvider, apiMux, agentFront))
 	t.Cleanup(ts.Close)
-	t.Cleanup(func() { svc.DrainConnections(time.Millisecond, true) })
+	t.Cleanup(func() { h.DrainConnections(time.Millisecond, true) })
 
 	// workers connect over WebTransport (control + data on one session)
-	udp, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0})
+	wtMux := http.NewServeMux()
+	wtMux.Handle("/agent", service.NewAgentWTService(h))
+	wt := service.NewWebTransportServer(selfSignedTLS(t))
+	wt.H3.Handler = service.NewWebTransportHandler(keyProvider, wt, wtMux)
+	bound, stopWT, err := service.ListenWebTransport(wt, []string{"127.0.0.1"}, 0)
 	require.NoError(t, err)
-	wt := service.NewAgentWebTransportServer(svc, keyProvider, selfSignedTLS(t))
-	go func() { _ = wt.Serve(udp) }()
-	t.Cleanup(func() { _ = wt.Close(); _ = udp.Close() })
-	wtURL := "https://" + udp.LocalAddr().String() + "/agent"
+	t.Cleanup(stopWT)
+	wtURL := "https://" + bound[0].String() + "/agent"
 
-	return &endpointStack{t: t, ts: ts, svc: svc, wtURL: wtURL}
+	return &endpointStack{t: t, ts: ts, handler: h, wtURL: wtURL}
 }
 
 func (s *endpointStack) startWorker(target string, deployment string, endpoints []*livekit.AgentHttp_AgentEndpoint) *conformance.Worker {

pkg/service/agenthandler.go

@@ -17,85 +17,29 @@ package service
 import (
 	"context"
 	"errors"
-	"fmt"
 	"math/rand"
-	"net/http"
 	"slices"
 	"sort"
-	"strconv"
 	"sync"
 	"time"
 
-	"github.com/gorilla/websocket"
 	"google.golang.org/protobuf/types/known/emptypb"
 
-	"github.com/livekit/livekit-server/pkg/agent"
-	"github.com/livekit/livekit-server/pkg/agent/endpoint"
-	"github.com/livekit/livekit-server/pkg/agent/endpoint/wire"
-	"github.com/livekit/livekit-server/pkg/config"
-	"github.com/livekit/livekit-server/pkg/routing"
-	"github.com/livekit/livekit-server/pkg/rtc"
-	"github.com/livekit/livekit-server/pkg/rtc/types"
-	"github.com/livekit/livekit-server/version"
 	"github.com/livekit/protocol/auth"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 	"github.com/livekit/protocol/rpc"
 	"github.com/livekit/protocol/utils"
 	"github.com/livekit/psrpc"
+
+	"github.com/livekit/livekit-server/pkg/agent"
+	"github.com/livekit/livekit-server/pkg/agent/endpoint"
+	"github.com/livekit/livekit-server/pkg/config"
+	"github.com/livekit/livekit-server/pkg/routing"
+	"github.com/livekit/livekit-server/pkg/rtc/types"
+	"github.com/livekit/livekit-server/version"
 )
 
-type AgentSocketUpgrader struct {
-	websocket.Upgrader
-}
-
-func (u AgentSocketUpgrader) Upgrade(
-	w http.ResponseWriter,
-	r *http.Request,
-	responseHeader http.Header,
-) (
-	conn *websocket.Conn,
-	registration agent.WorkerRegistration,
-	ok bool,
-) {
-	if u.CheckOrigin == nil {
-		// allow connections from any origin, since script may be hosted anywhere
-		// security is enforced by access tokens
-		u.CheckOrigin = func(r *http.Request) bool {
-			return true
-		}
-	}
-
-	// reject non websocket requests
-	if !websocket.IsWebSocketUpgrade(r) {
-		w.WriteHeader(404)
-		return
-	}
-
-	// require a claim
-	claims := GetGrants(r.Context())
-	if claims == nil || claims.Video == nil || !claims.Video.Agent {
-		HandleError(w, r, http.StatusUnauthorized, rtc.ErrPermissionDenied)
-		return
-	}
-
-	registration = agent.MakeWorkerRegistration()
-	registration.ClientIP = GetClientIP(r)
-
-	// upgrade
-	conn, err := u.Upgrader.Upgrade(w, r, responseHeader)
-	if err != nil {
-		HandleError(w, r, http.StatusInternalServerError, err)
-		return
-	}
-
-	if pv, err := strconv.Atoi(r.FormValue("protocol")); err == nil {
-		registration.Protocol = agent.WorkerProtocolVersion(pv)
-	}
-
-	return conn, registration, true
-}
-
 func DispatchAgentWorkerSignal(c agent.SignalConn, h agent.WorkerSignalHandler, l logger.Logger) bool {
 	req, _, err := c.ReadWorkerMessage()
 	if err != nil {
@@ -115,8 +59,8 @@ func DispatchAgentWorkerSignal(c agent.SignalConn, h agent.WorkerSignalHandler,
 	return true
 }
 
-func HandshakeAgentWorker(c agent.SignalConn, serverInfo *livekit.ServerInfo, registration agent.WorkerRegistration, l logger.Logger, endpointSettings agent.EndpointSettingsFunc) (r agent.WorkerRegistration, ok bool) {
-	wr := agent.NewWorkerRegisterer(c, serverInfo, registration, endpointSettings)
+func HandshakeAgentWorker(c agent.SignalConn, serverInfo *livekit.ServerInfo, registration agent.WorkerRegistration, l logger.Logger, handlers ...agent.WorkerRegisterHandler) (r agent.WorkerRegistration, ok bool) {
+	wr := agent.NewWorkerRegisterer(c, serverInfo, registration, handlers...)
 	if err := c.SetReadDeadline(wr.Deadline()); err != nil {
 		return
 	}
@@ -131,17 +75,6 @@ func HandshakeAgentWorker(c agent.SignalConn, serverInfo *livekit.ServerInfo, re
 	return wr.Registration(), true
 }
 
-type AgentService struct {
-	upgrader AgentSocketUpgrader
-
-	signalMessageSizeLimit int64
-	// developmentMode allows the WebTransport listener to run without a
-	// configured certificate.
-	developmentMode bool
-
-	*AgentHandler
-}
-
 type AgentHandler struct {
 	agentServer rpc.AgentInternalServer
 	// the server's only configured api key, when exactly one exists: the
@@ -179,138 +112,58 @@ type workerKey struct {
 	deployment string
 }
 
-func NewAgentService(
+// NewAgentHandler builds the node's agent worker handler, shared by the
+// transport services.
+func NewAgentHandler(
 	conf *config.Config,
 	currentNode routing.LocalNode,
 	bus psrpc.MessageBus,
 	keyProvider auth.KeyProvider,
-) (*AgentService, error) {
-	s := &AgentService{
-		signalMessageSizeLimit: conf.Limit.AgentSignalMessageSizeLimit,
-		developmentMode:        conf.Development,
-	}
-
-	serverInfo := &livekit.ServerInfo{
-		Edition:       livekit.ServerInfo_Standard,
-		Version:       version.Version,
-		Protocol:      types.CurrentProtocol,
-		AgentProtocol: agent.CurrentProtocol,
-		Region:        conf.Region,
-		NodeId:        string(currentNode.NodeID()),
-	}
-
-	agentServer, err := rpc.NewAgentInternalServer(s, bus)
-	if err != nil {
-		return nil, err
-	}
-	s.AgentHandler = NewAgentHandler(
-		agentServer,
-		keyProvider,
-		logger.GetLogger(),
-		serverInfo,
-		conf.Agents.TargetLoad,
-		agent.RoomAgentTopic,
-		agent.PublisherAgentTopic,
-		agent.ParticipantAgentTopic,
-	)
-	s.AgentHandler.endpointsConfig = conf.Agents.Endpoints
-	if len(conf.Keys) == 1 {
-		for key := range conf.Keys {
-			s.AgentHandler.singleAPIKey = key
-		}
-	}
-
-	return s, nil
-}
-
-// EndpointFront is the /agents/{agent_name}/{deployment}/{path...} handler backed
-// by this node's attached workers. The api key comes from validated grants when a
-// token is present; a non-public route additionally requires an agent-endpoint
-// grant scoped to this agent and deployment.
-func (s *AgentService) EndpointFront() http.Handler {
-	return endpoint.NewFront(endpoint.FrontParams{
-		Registry: s.endpointRegistry,
-		ResolveAccess: func(r *http.Request, agentName, deployment string) endpoint.Access {
-			if claims := GetGrants(r.Context()); claims != nil {
-				level := endpoint.AccessCredentialed
-				if claims.AgentEndpoint.Allows(agentName, deployment) {
-					level = endpoint.AccessGranted
-				}
-				return endpoint.Access{APIKey: GetAPIKey(r.Context()), Level: level}
-			}
-			// unauthenticated: one configured key makes the api key unambiguous
-			return endpoint.Access{APIKey: s.singleAPIKey}
-		},
-		Logger:            s.logger,
-		SingleKeyFallback: true,
-	})
-}
-
-func (s *AgentService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	if conn, registration, ok := s.upgrader.Upgrade(w, r, nil); ok {
-		// control/signal conn: bound a single signalling frame (0 = operator
-		// disabled) before it is buffered; decompressed size is bounded in
-		// WSSignalConnection. HTTP endpoints are served over a worker's
-		// WebTransport session (see the WebTransport listener), not this
-		// WebSocket, which now carries control/job-dispatch traffic only.
-		if s.signalMessageSizeLimit > 0 {
-			conn.SetReadLimit(s.signalMessageSizeLimit)
-		}
-		sigConn := NewWSSignalConnection(conn, s.signalMessageSizeLimit)
-		defer sigConn.Close()
-		s.HandleConnection(r.Context(), sigConn, registration)
-	}
-}
-
-func NewAgentHandler(
-	agentServer rpc.AgentInternalServer,
-	keyProvider auth.KeyProvider,
-	logger logger.Logger,
-	serverInfo *livekit.ServerInfo,
-	targetLoad float32,
-	roomTopic string,
-	publisherTopic string,
-	participantTopic string,
-) *AgentHandler {
-	return &AgentHandler{
-		agentServer:      agentServer,
-		logger:           logger.WithComponent("agents"),
+	registry *endpoint.Registry,
+) (*AgentHandler, error) {
+	h := &AgentHandler{
+		logger:           logger.GetLogger().WithComponent("agents"),
 		workers:          make(map[string]*agent.Worker),
 		jobToWorker:      make(map[livekit.JobID]*agent.Worker),
 		namespaceWorkers: make(map[workerKey][]*agent.Worker),
-		serverInfo:       serverInfo,
+		serverInfo: &livekit.ServerInfo{
+			Edition:       livekit.ServerInfo_Standard,
+			Version:       version.Version,
+			Protocol:      types.CurrentProtocol,
+			AgentProtocol: agent.CurrentProtocol,
+			Region:        conf.Region,
+			NodeId:        string(currentNode.NodeID()),
+		},
 		keyProvider:      keyProvider,
-		targetLoad:       targetLoad,
-		roomTopic:        roomTopic,
-		publisherTopic:   publisherTopic,
-		participantTopic: participantTopic,
-		endpointRegistry: endpoint.NewRegistry(),
+		targetLoad:       conf.Agents.TargetLoad,
+		roomTopic:        agent.RoomAgentTopic,
+		publisherTopic:   agent.PublisherAgentTopic,
+		participantTopic: agent.ParticipantAgentTopic,
+		endpointRegistry: registry,
+		endpointsConfig:  conf.Agents.Endpoints,
+	}
+	if len(conf.Keys) == 1 {
+		for key := range conf.Keys {
+			h.singleAPIKey = key
+		}
 	}
-}
 
-// endpointSettings validates a registration's manifest and negotiates the
-// data-plane protocol version.
-func (h *AgentHandler) endpointSettings(req *livekit.RegisterWorkerRequest) (*livekit.AgentHttp_AgentEndpointSettings, error) {
-	if h.endpointsConfig.Disabled {
-		return nil, errors.New("agent HTTP endpoints are disabled on this server")
-	}
-	if _, err := endpoint.ParseManifest(req.GetEndpoints()); err != nil {
+	agentServer, err := rpc.NewAgentInternalServer(h, bus)
+	if err != nil {
 		return nil, err
 	}
-	if req.GetInstanceId() == "" {
-		return nil, errors.New("registrations with endpoints require an instance_id")
+	h.agentServer = agentServer
+
+	return h, nil
+}
+
+// endpointRegisterHandler negotiates a worker's endpoint settings, refusing any
+// declaration when endpoints are turned off.
+func (h *AgentHandler) endpointRegisterHandler(req *livekit.RegisterWorkerRequest, res *livekit.RegisterWorkerResponse, reg *agent.WorkerRegistration) error {
+	if len(req.GetEndpoints()) > 0 && h.endpointsConfig.Disabled {
+		return errors.New("agent HTTP endpoints are disabled on this server")
 	}
-	if req.GetEndpointProtocol() == 0 {
-		return nil, errors.New("worker declared endpoints but no endpoint protocol; upgrade the agent SDK to one that speaks the endpoint data plane")
-	}
-	// the worker frames to the version returned here, so it must be the
-	// negotiated one and not the server's own constant
-	negotiated := min(req.GetEndpointProtocol(), wire.CurrentProtocol)
-	if negotiated < wire.MinProtocol {
-		return nil, fmt.Errorf("unsupported agent endpoint protocol %d (this server serves %d..%d)",
-			req.GetEndpointProtocol(), wire.MinProtocol, wire.CurrentProtocol)
-	}
-	return &livekit.AgentHttp_AgentEndpointSettings{Protocol: negotiated}, nil
+	return agent.EndpointRegisterHandler(req, res, reg)
 }
 
 // HandleConnection serves a worker's control connection with no data-plane
@@ -324,7 +177,7 @@ func (h *AgentHandler) HandleConnection(ctx context.Context, conn agent.SignalCo
 // exchanges share it); nil for a WebSocket control connection, which serves no
 // endpoints.
 func (h *AgentHandler) handleConnection(ctx context.Context, conn agent.SignalConn, registration agent.WorkerRegistration, sess endpoint.Session) {
-	registration, ok := HandshakeAgentWorker(conn, h.serverInfo, registration, h.logger, h.endpointSettings)
+	registration, ok := HandshakeAgentWorker(conn, h.serverInfo, registration, h.logger, h.endpointRegisterHandler)
 	if !ok {
 		return
 	}
@@ -513,6 +366,7 @@ func (h *AgentHandler) JobRequest(ctx context.Context, job *livekit.Job) (*rpc.J
 		"jobID", job.Id,
 		"namespace", job.Namespace,
 		"agentName", job.AgentName,
+		"deployment", job.Deployment,
 		"jobType", job.Type.String(),
 	)
 	if job.Room != nil {

pkg/service/agentwebtransport.go

@@ -1,230 +0,0 @@
-// Copyright 2026 LiveKit, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package service
-
-import (
-	"context"
-	"crypto/ecdsa"
-	"crypto/elliptic"
-	"crypto/rand"
-	"crypto/tls"
-	"crypto/x509"
-	"crypto/x509/pkix"
-	"errors"
-	"math/big"
-	"net"
-	"net/http"
-	"strconv"
-	"sync"
-	"time"
-
-	"github.com/quic-go/quic-go/http3"
-	"github.com/quic-go/webtransport-go"
-
-	"github.com/livekit/livekit-server/pkg/agent"
-	"github.com/livekit/livekit-server/pkg/agent/endpoint"
-	"github.com/livekit/livekit-server/pkg/agent/endpoint/wire"
-	"github.com/livekit/livekit-server/pkg/rtc"
-	"github.com/livekit/protocol/auth"
-	"github.com/livekit/protocol/livekit"
-	"github.com/livekit/protocol/logger"
-)
-
-// StartWebTransport starts the worker WebTransport listener (control + data) on
-// the configured UDP port, if endpoints are enabled and a port is set. It
-// returns the stop function, nil when no listener was started. QUIC has no
-// plaintext mode, so a TLS certificate is required: from
-// tls_cert_file/tls_key_file, or a generated self-signed cert in dev mode.
-func (s *AgentService) StartWebTransport() (func(), error) {
-	cfg := s.endpointsConfig
-	if cfg.Disabled || cfg.WebTransportPort == 0 {
-		return nil, nil
-	}
-	tlsConf, err := WebTransportTLS(cfg.TLSCertFile, cfg.TLSKeyFile, s.developmentMode)
-	if err != nil {
-		return nil, err
-	}
-	udp, err := net.ListenUDP("udp", &net.UDPAddr{Port: int(cfg.WebTransportPort)})
-	if err != nil {
-		return nil, err
-	}
-	wt := NewAgentWebTransportServer(s, s.keyProvider, tlsConf)
-	go func() {
-		if err := wt.Serve(udp); err != nil {
-			logger.Infow("agent webtransport listener stopped", "error", err)
-		}
-	}()
-	logger.Infow("agent webtransport listener started", "port", cfg.WebTransportPort)
-	return func() { _ = wt.Close(); _ = udp.Close() }, nil
-}
-
-// WebTransportTLS builds the listener's TLS config from cert files, or a
-// generated self-signed cert in dev mode.
-func WebTransportTLS(certFile, keyFile string, dev bool) (*tls.Config, error) {
-	if certFile != "" && keyFile != "" {
-		cert, err := tls.LoadX509KeyPair(certFile, keyFile)
-		if err != nil {
-			return nil, err
-		}
-		return &tls.Config{Certificates: []tls.Certificate{cert}, NextProtos: []string{http3.NextProtoH3}}, nil
-	}
-	if !dev {
-		return nil, errors.New("agent endpoints WebTransport requires agents.endpoints.tls_cert_file/tls_key_file")
-	}
-	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
-	if err != nil {
-		return nil, err
-	}
-	tmpl := &x509.Certificate{
-		SerialNumber: big.NewInt(1),
-		Subject:      pkix.Name{CommonName: "livekit-agent-webtransport-dev"},
-		NotBefore:    time.Now().Add(-time.Hour),
-		NotAfter:     time.Now().Add(365 * 24 * time.Hour),
-		IPAddresses:  []net.IP{net.IPv4(127, 0, 0, 1), net.IPv6loopback},
-		DNSNames:     []string{"localhost"},
-	}
-	der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
-	if err != nil {
-		return nil, err
-	}
-	logger.Warnw("agent webtransport using a generated self-signed cert (dev mode); workers must skip verification", nil)
-	return &tls.Config{
-		Certificates: []tls.Certificate{{Certificate: [][]byte{der}, PrivateKey: key}},
-		NextProtos:   []string{http3.NextProtoH3},
-	}, nil
-}
-
-// NewWebTransportServer wraps an HTTP/3 WebTransport server around tlsConf (the
-// h3 ALPN is set here if absent). The caller must assign wt.H3.Handler: routes
-// that Upgrade need the *webtransport.Server itself.
-func NewWebTransportServer(tlsConf *tls.Config) *webtransport.Server {
-	tlsConf = tlsConf.Clone()
-	if len(tlsConf.NextProtos) == 0 {
-		tlsConf.NextProtos = []string{http3.NextProtoH3}
-	}
-	return &webtransport.Server{H3: &http3.Server{TLSConfig: tlsConf}}
-}
-
-// NewAgentWebTransportServer builds the WebTransport server that terminates a
-// worker's unified session on /agent: the control stream carries the same
-// WorkerMessage/ServerMessage exchange as the WebSocket control connection, and
-// each HTTP exchange rides a node-opened QUIC stream on the same session. The
-// handler runs behind the same api-key auth middleware as the rest of the
-// agent surface, so the agent grant is enforced identically.
-func NewAgentWebTransportServer(svc *AgentService, keyProvider auth.KeyProvider, tlsConf *tls.Config) *webtransport.Server {
-	wt := NewWebTransportServer(tlsConf)
-	authMW := NewAPIKeyAuthMiddleware(keyProvider)
-	mux := http.NewServeMux()
-	mux.HandleFunc("/agent", func(w http.ResponseWriter, r *http.Request) {
-		svc.ServeWebTransport(wt, w, r)
-	})
-	wt.H3.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		authMW.ServeHTTP(w, r, mux.ServeHTTP)
-	})
-	return wt
-}
-
-// ServeWebTransport verifies the agent grant, upgrades the request to a
-// WebTransport session, and serves it. It assumes an outer auth middleware has
-// already placed the grant in the request context (as the WebSocket path does).
-func (s *AgentService) ServeWebTransport(wt *webtransport.Server, w http.ResponseWriter, r *http.Request) {
-	claims := GetGrants(r.Context())
-	if claims == nil || claims.Video == nil || !claims.Video.Agent {
-		HandleError(w, r, http.StatusUnauthorized, rtc.ErrPermissionDenied)
-		return
-	}
-	apiKey := GetAPIKey(r.Context())
-
-	registration := agent.MakeWorkerRegistration()
-	registration.ClientIP = GetClientIP(r)
-	if pv, err := strconv.Atoi(r.FormValue("protocol")); err == nil {
-		registration.Protocol = agent.WorkerProtocolVersion(pv)
-	}
-
-	sess, err := wt.Upgrade(w, r)
-	if err != nil {
-		s.logger.Warnw("agent webtransport upgrade failed", err)
-		return
-	}
-
-	// a detached context carries the worker's identity for the whole session;
-	// the control loop exits on the control stream's read error when the session
-	// dies, so it needs no cancellation from the request.
-	ctx := WithGrants(context.Background(), claims, apiKey)
-	go s.serveWebTransportSession(ctx, sess, registration)
-}
-
-func (s *AgentService) serveWebTransportSession(ctx context.Context, sess *webtransport.Session, registration agent.WorkerRegistration) {
-	// the worker opens the control stream first (see the conformance worker and
-	// the SDK); every other stream the node opens is an HTTP exchange. Bound the
-	// wait so a session that upgrades but never opens a control stream can't hold
-	// a goroutine until the QUIC idle timeout.
-	acceptCtx, cancel := context.WithTimeout(sess.Context(), agent.RegisterTimeout)
-	defer cancel()
-	control, err := sess.AcceptStream(acceptCtx)
-	if err != nil {
-		_ = sess.CloseWithError(wire.SessionCloseOK, "no control stream")
-		return
-	}
-	sigConn := NewWTSignalConn(sess, control)
-	defer sigConn.Close()
-	maxStreams := int(s.endpointsConfig.MaxStreams)
-	if maxStreams <= 0 {
-		maxStreams = endpoint.DefaultMaxStreams
-	}
-	epSession := endpoint.NewWebTransportSession(sess, maxStreams)
-	s.handleConnection(ctx, sigConn, registration, epSession)
-}
-
-// wtSignalConn adapts a WebTransport control stream to agent.SignalConn: the
-// same length-delimited WorkerMessage/ServerMessage exchange the WebSocket
-// control connection carries, framed on one QUIC bidirectional stream.
-type wtSignalConn struct {
-	sess    *webtransport.Session
-	control *webtransport.Stream
-	writeMu sync.Mutex
-}
-
-// NewWTSignalConn adapts a WebTransport session's control stream to
-// agent.SignalConn.
-func NewWTSignalConn(sess *webtransport.Session, control *webtransport.Stream) agent.SignalConn {
-	return &wtSignalConn{sess: sess, control: control}
-}
-
-func (c *wtSignalConn) WriteServerMessage(msg *livekit.ServerMessage) (int, error) {
-	c.writeMu.Lock()
-	defer c.writeMu.Unlock()
-	return 0, wire.WriteControlMessage(c.control, msg)
-}
-
-func (c *wtSignalConn) ReadWorkerMessage() (*livekit.WorkerMessage, int, error) {
-	var msg livekit.WorkerMessage
-	if err := wire.ReadControlMessage(c.control, &msg); err != nil {
-		return nil, 0, err
-	}
-	return &msg, 0, nil
-}
-
-func (c *wtSignalConn) SetReadDeadline(t time.Time) error {
-	return c.control.SetReadDeadline(t)
-}
-
-func (c *wtSignalConn) Close() error {
-	return c.sess.CloseWithError(wire.SessionCloseOK, "")
-}
-
-func (c *wtSignalConn) CloseWithReason(reason string) error {
-	return c.sess.CloseWithError(wire.SessionCloseOK, reason)
-}

pkg/service/agentws.go

@@ -0,0 +1,102 @@
+// Copyright 2024 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package service
+
+import (
+	"net/http"
+	"strconv"
+
+	"github.com/gorilla/websocket"
+
+	"github.com/livekit/livekit-server/pkg/agent"
+	"github.com/livekit/livekit-server/pkg/config"
+	"github.com/livekit/livekit-server/pkg/rtc"
+)
+
+type AgentSocketUpgrader struct {
+	websocket.Upgrader
+}
+
+func (u AgentSocketUpgrader) Upgrade(
+	w http.ResponseWriter,
+	r *http.Request,
+	responseHeader http.Header,
+) (
+	conn *websocket.Conn,
+	registration agent.WorkerRegistration,
+	ok bool,
+) {
+	if u.CheckOrigin == nil {
+		// allow connections from any origin, since script may be hosted anywhere
+		// security is enforced by access tokens
+		u.CheckOrigin = func(r *http.Request) bool {
+			return true
+		}
+	}
+
+	if !websocket.IsWebSocketUpgrade(r) {
+		w.WriteHeader(404)
+		return
+	}
+
+	claims := GetGrants(r.Context())
+	if claims == nil || claims.Video == nil || !claims.Video.Agent {
+		HandleError(w, r, http.StatusUnauthorized, rtc.ErrPermissionDenied)
+		return
+	}
+
+	registration = agent.MakeWorkerRegistration()
+	registration.ClientIP = GetClientIP(r)
+
+	conn, err := u.Upgrader.Upgrade(w, r, responseHeader)
+	if err != nil {
+		HandleError(w, r, http.StatusInternalServerError, err)
+		return
+	}
+
+	if pv, err := strconv.Atoi(r.FormValue("protocol")); err == nil {
+		registration.Protocol = agent.WorkerProtocolVersion(pv)
+	}
+
+	return conn, registration, true
+}
+
+// AgentWSService serves a worker's WebSocket control connection on /agent:
+// control and job dispatch only, with no data-plane session for HTTP endpoints.
+type AgentWSService struct {
+	*AgentHandler
+
+	upgrader               AgentSocketUpgrader
+	signalMessageSizeLimit int64
+}
+
+func NewAgentWSService(conf *config.Config, h *AgentHandler) *AgentWSService {
+	return &AgentWSService{
+		AgentHandler:           h,
+		signalMessageSizeLimit: conf.Limit.AgentSignalMessageSizeLimit,
+	}
+}
+
+func (s *AgentWSService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if conn, registration, ok := s.upgrader.Upgrade(w, r, nil); ok {
+		// bound a single signalling frame before it is buffered; 0 disables
+		if s.signalMessageSizeLimit > 0 {
+			conn.SetReadLimit(s.signalMessageSizeLimit)
+		}
+		sigConn := NewWSSignalConnection(conn, s.signalMessageSizeLimit)
+		defer sigConn.Close()
+		s.HandleConnection(r.Context(), sigConn, registration)
+	}
+}

pkg/service/agentwt.go

@@ -0,0 +1,143 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package service
+
+import (
+	"context"
+	"net/http"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/quic-go/webtransport-go"
+
+	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/logger"
+
+	"github.com/livekit/livekit-server/pkg/agent"
+	"github.com/livekit/livekit-server/pkg/agent/endpoint"
+	"github.com/livekit/livekit-server/pkg/agent/endpoint/wire"
+	"github.com/livekit/livekit-server/pkg/rtc"
+)
+
+// AgentWTService serves a worker's WebTransport session on /agent: one control
+// stream, plus a node-opened QUIC stream per HTTP exchange.
+type AgentWTService struct {
+	*AgentHandler
+}
+
+func NewAgentWTService(h *AgentHandler) *AgentWTService {
+	return &AgentWTService{AgentHandler: h}
+}
+
+func (s *AgentWTService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	sess, registration, ok := UpgradeAgentWebTransport(w, r, s.logger)
+	if !ok {
+		return
+	}
+
+	sigConn, epSession, ok := AcceptAgentControlStream(sess, s.endpointsConfig.MaxStreams)
+	if !ok {
+		return
+	}
+	defer sigConn.Close()
+
+	s.handleConnection(r.Context(), sigConn, registration, epSession)
+}
+
+// UpgradeAgentWebTransport verifies the agent grant, builds the worker's
+// registration from the request and upgrades to a WebTransport session. An outer
+// auth middleware must already have placed the grant in the request context.
+func UpgradeAgentWebTransport(w http.ResponseWriter, r *http.Request, l logger.Logger) (
+	sess *webtransport.Session,
+	registration agent.WorkerRegistration,
+	ok bool,
+) {
+	claims := GetGrants(r.Context())
+	if claims == nil || claims.Video == nil || !claims.Video.Agent {
+		HandleError(w, r, http.StatusUnauthorized, rtc.ErrPermissionDenied)
+		return
+	}
+
+	registration = agent.MakeWorkerRegistration()
+	registration.ClientIP = GetClientIP(r)
+	if pv, err := strconv.Atoi(r.FormValue("protocol")); err == nil {
+		registration.Protocol = agent.WorkerProtocolVersion(pv)
+	}
+
+	sess, err := UpgradeWebTransport(w, r)
+	if err != nil {
+		l.Warnw("agent webtransport upgrade failed", err)
+		return nil, registration, false
+	}
+
+	return sess, registration, true
+}
+
+// AcceptAgentControlStream waits for the worker's control stream and wraps the
+// session for the data plane. The wait is bounded so a session that upgrades but
+// never opens a control stream can't hold a goroutine until the QUIC idle timeout.
+func AcceptAgentControlStream(sess *webtransport.Session, maxStreams uint32) (agent.SignalConn, endpoint.Session, bool) {
+	acceptCtx, cancel := context.WithTimeout(sess.Context(), agent.RegisterTimeout)
+	defer cancel()
+	control, err := sess.AcceptStream(acceptCtx)
+	if err != nil {
+		_ = sess.CloseWithError(wire.SessionCloseOK, "no control stream")
+		return nil, nil, false
+	}
+	if maxStreams == 0 {
+		maxStreams = endpoint.DefaultMaxStreams
+	}
+	return NewWTSignalConn(sess, control), endpoint.NewWebTransportSession(sess, int(maxStreams)), true
+}
+
+// wtSignalConn adapts a WebTransport control stream to agent.SignalConn: a
+// length-delimited WorkerMessage/ServerMessage exchange on one QUIC bidirectional
+// stream.
+type wtSignalConn struct {
+	sess    *webtransport.Session
+	control *webtransport.Stream
+	writeMu sync.Mutex
+}
+
+func NewWTSignalConn(sess *webtransport.Session, control *webtransport.Stream) agent.SignalConn {
+	return &wtSignalConn{sess: sess, control: control}
+}
+
+func (c *wtSignalConn) WriteServerMessage(msg *livekit.ServerMessage) (int, error) {
+	c.writeMu.Lock()
+	defer c.writeMu.Unlock()
+	return 0, wire.WriteControlMessage(c.control, msg)
+}
+
+func (c *wtSignalConn) ReadWorkerMessage() (*livekit.WorkerMessage, int, error) {
+	var msg livekit.WorkerMessage
+	if err := wire.ReadControlMessage(c.control, &msg); err != nil {
+		return nil, 0, err
+	}
+	return &msg, 0, nil
+}
+
+func (c *wtSignalConn) SetReadDeadline(t time.Time) error {
+	return c.control.SetReadDeadline(t)
+}
+
+func (c *wtSignalConn) Close() error {
+	return c.sess.CloseWithError(wire.SessionCloseOK, "")
+}
+
+func (c *wtSignalConn) CloseWithReason(reason string) error {
+	return c.sess.CloseWithError(wire.SessionCloseOK, reason)
+}

pkg/service/server.go

@@ -16,6 +16,7 @@ package service
 
 import (
 	"context"
+	"crypto/tls"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -29,6 +30,7 @@ import (
 
 	"github.com/pion/turn/v5"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/quic-go/webtransport-go"
 	"github.com/rs/cors"
 	"github.com/twitchtv/twirp"
 	"github.com/urfave/negroni/v3"
@@ -46,23 +48,22 @@ import (
 )
 
 type LivekitServer struct {
-	config       *config.Config
-	ioService    *IOInfoService
-	rtcService   *RTCService
-	whipService  *WHIPService
-	agentService *AgentService
-	httpServer   *http.Server
-	promServer   *http.Server
-	debugServer  *http.Server
-	router       routing.Router
-	roomManager  *RoomManager
-	signalServer *SignalServer
-	turnServer   *turn.Server
-	currentNode  routing.LocalNode
-	agentWTStop  func()
-	running      atomic.Bool
-	doneChan     chan struct{}
-	closedChan   chan struct{}
+	config             *config.Config
+	ioService          *IOInfoService
+	rtcService         *RTCService
+	whipService        *WHIPService
+	httpServer         *http.Server
+	promServer         *http.Server
+	debugServer        *http.Server
+	webtransportServer *webtransport.Server
+	router             routing.Router
+	roomManager        *RoomManager
+	signalServer       *SignalServer
+	turnServer         *turn.Server
+	currentNode        routing.LocalNode
+	running            atomic.Bool
+	doneChan           chan struct{}
+	closedChan         chan struct{}
 }
 
 func NewLivekitServer(conf *config.Config,
@@ -74,7 +75,9 @@ func NewLivekitServer(conf *config.Config,
 	ioService *IOInfoService,
 	rtcService *RTCService,
 	whipService *WHIPService,
-	agentService *AgentService,
+	agentWSService *AgentWSService,
+	agentWTService *AgentWTService,
+	agentEndpointService *AgentEndpointService,
 	keyProvider auth.KeyProvider,
 	router routing.Router,
 	roomManager *RoomManager,
@@ -87,7 +90,6 @@ func NewLivekitServer(conf *config.Config,
 		ioService:    ioService,
 		rtcService:   rtcService,
 		whipService:  whipService,
-		agentService: agentService,
 		router:       router,
 		roomManager:  roomManager,
 		signalServer: signalServer,
@@ -128,12 +130,13 @@ func NewLivekitServer(conf *config.Config,
 	xtwirp.RegisterServer(mux, sipServer)
 	rtcService.SetupRoutes(mux)
 	whipService.SetupRoutes(mux)
-	mux.Handle("/agent", agentService)
+	mux.Handle("/agent", agentWSService)
 	mux.HandleFunc("/", s.defaultHandler)
 
+	// NewHTTPHandler branches on a nil handler, which a typed-nil would defeat
 	var agentFront http.Handler
 	if !conf.Agents.Endpoints.Disabled {
-		agentFront = agentService.EndpointFront()
+		agentFront = agentEndpointService
 	}
 
 	s.httpServer = &http.Server{
@@ -176,6 +179,19 @@ func NewLivekitServer(conf *config.Config,
 		}
 	}
 
+	if conf.WebTransport.Port > 0 {
+		var tlsConf *tls.Config
+		tlsConf, err = WebTransportTLS(conf.WebTransport.TLSCertFile, conf.WebTransport.TLSKeyFile, conf.Development)
+		if err != nil {
+			return
+		}
+		wtMux := http.NewServeMux()
+		wtMux.Handle("/agent", agentWTService)
+
+		s.webtransportServer = NewWebTransportServer(tlsConf)
+		s.webtransportServer.H3.Handler = NewWebTransportHandler(keyProvider, s.webtransportServer, wtMux)
+	}
+
 	if err = router.RemoveDeadNodes(); err != nil {
 		return
 	}
@@ -218,14 +234,6 @@ func (s *LivekitServer) Start() error {
 		return err
 	}
 
-	if s.agentService != nil {
-		stop, err := s.agentService.StartWebTransport()
-		if err != nil {
-			return err
-		}
-		s.agentWTStop = stop
-	}
-
 	addresses := s.config.BindAddresses
 	if addresses == nil {
 		addresses = []string{""}
@@ -259,6 +267,15 @@ func (s *LivekitServer) Start() error {
 		}
 	}
 
+	stopWebTransport := func() {}
+	if s.webtransportServer != nil {
+		_, stop, err := ListenWebTransport(s.webtransportServer, s.config.BindAddresses, s.config.WebTransport.Port)
+		if err != nil {
+			return err
+		}
+		stopWebTransport = stop
+	}
+
 	values := []any{
 		"portHttp", s.config.Port,
 		"nodeID", s.currentNode.NodeID(),
@@ -284,6 +301,9 @@ func (s *LivekitServer) Start() error {
 	if s.config.DebugHandler.Port != 0 {
 		values = append(values, "portDebugHandler", s.config.DebugHandler.Port)
 	}
+	if s.config.WebTransport.Port != 0 {
+		values = append(values, "portWebTransport", s.config.WebTransport.Port)
+	}
 	if s.config.Region != "" {
 		values = append(values, "region", s.config.Region)
 	}
@@ -334,6 +354,7 @@ func (s *LivekitServer) Start() error {
 	if s.debugServer != nil {
 		_ = s.debugServer.Shutdown(ctx)
 	}
+	stopWebTransport()
 
 	if s.turnServer != nil {
 		_ = s.turnServer.Close()
@@ -364,9 +385,6 @@ func (s *LivekitServer) Stop(force bool) {
 	}
 
 	s.router.Stop()
-	if s.agentWTStop != nil {
-		s.agentWTStop()
-	}
 	close(s.doneChan)
 
 	// wait for fully closed

pkg/service/webtransport.go

@@ -0,0 +1,177 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package service
+
+import (
+	"context"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"errors"
+	"math/big"
+	"net"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/quic-go/quic-go/http3"
+	"github.com/quic-go/webtransport-go"
+	"github.com/urfave/negroni/v3"
+
+	"github.com/livekit/protocol/auth"
+	"github.com/livekit/protocol/logger"
+)
+
+// WebTransportTLS builds the listener's TLS config from cert files, or a
+// generated self-signed cert in dev mode.
+func WebTransportTLS(certFile, keyFile string, dev bool) (*tls.Config, error) {
+	if certFile != "" && keyFile != "" {
+		cert, err := tls.LoadX509KeyPair(certFile, keyFile)
+		if err != nil {
+			return nil, err
+		}
+		return &tls.Config{Certificates: []tls.Certificate{cert}, NextProtos: []string{http3.NextProtoH3}}, nil
+	}
+	if !dev {
+		return nil, errors.New("webtransport requires webtransport.tls_cert_file/tls_key_file")
+	}
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		return nil, err
+	}
+	tmpl := &x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		Subject:      pkix.Name{CommonName: "livekit-webtransport-dev"},
+		NotBefore:    time.Now().Add(-time.Hour),
+		NotAfter:     time.Now().Add(365 * 24 * time.Hour),
+		IPAddresses:  []net.IP{net.IPv4(127, 0, 0, 1), net.IPv6loopback},
+		DNSNames:     []string{"localhost"},
+	}
+	der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
+	if err != nil {
+		return nil, err
+	}
+	logger.Warnw("webtransport using a generated self-signed cert (dev mode); clients must skip verification", nil)
+	return &tls.Config{
+		Certificates: []tls.Certificate{{Certificate: [][]byte{der}, PrivateKey: key}},
+		NextProtos:   []string{http3.NextProtoH3},
+	}, nil
+}
+
+// NewWebTransportServer wraps an HTTP/3 WebTransport server around tlsConf (the
+// h3 ALPN is set here if absent). The caller must assign wt.H3.Handler.
+func NewWebTransportServer(tlsConf *tls.Config) *webtransport.Server {
+	tlsConf = tlsConf.Clone()
+	if len(tlsConf.NextProtos) == 0 {
+		tlsConf.NextProtos = []string{http3.NextProtoH3}
+	}
+	return &webtransport.Server{H3: &http3.Server{TLSConfig: tlsConf}}
+}
+
+// NewWebTransportHandler builds the listener's handler: mux behind api-key auth,
+// with wt in each request's context.
+func NewWebTransportHandler(keyProvider auth.KeyProvider, wt *webtransport.Server, mux http.Handler) http.Handler {
+	middlewares := []negroni.Handler{negroni.NewRecovery()}
+	if keyProvider != nil {
+		middlewares = append(middlewares, NewAPIKeyAuthMiddleware(keyProvider))
+	}
+	return WithWebTransportServer(wt, WithPathNormalization(configureMiddlewares(mux, middlewares...)))
+}
+
+type webTransportServerKey struct{}
+
+// WithWebTransportServer puts wt in each request's context, where a route that
+// upgrades reads it. Apply it outermost, ahead of the listener's middleware chain.
+func WithWebTransportServer(wt *webtransport.Server, next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		next.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), webTransportServerKey{}, wt)))
+	})
+}
+
+// GetWebTransportServer returns the server serving this request, or nil when the
+// request did not arrive over a WebTransport listener.
+func GetWebTransportServer(ctx context.Context) *webtransport.Server {
+	wt, _ := ctx.Value(webTransportServerKey{}).(*webtransport.Server)
+	return wt
+}
+
+// UpgradeWebTransport upgrades a request on the server carried in its context.
+func UpgradeWebTransport(w http.ResponseWriter, r *http.Request) (*webtransport.Session, error) {
+	wt := GetWebTransportServer(r.Context())
+	if wt == nil {
+		return nil, errors.New("no webtransport server serving this request")
+	}
+	return wt.Upgrade(unwrapResponseWriter(w), r)
+}
+
+// unwrapResponseWriter reaches the ResponseWriter net/http handed in. An upgrade
+// type-asserts it to http3.Settingser and http3.HTTPStreamer without checking, so
+// it must not be a middleware's wrapper.
+func unwrapResponseWriter(w http.ResponseWriter) http.ResponseWriter {
+	for {
+		u, ok := w.(interface{ Unwrap() http.ResponseWriter })
+		if !ok {
+			return w
+		}
+		w = u.Unwrap()
+	}
+}
+
+// ListenWebTransport binds a UDP socket per address and serves wt on each,
+// returning the bound addresses and a stop func. Empty addrs binds all interfaces.
+func ListenWebTransport(wt *webtransport.Server, addrs []string, port uint32) ([]net.Addr, func(), error) {
+	if len(addrs) == 0 {
+		addrs = []string{""}
+	}
+
+	conns := make([]*net.UDPConn, 0, len(addrs))
+	bound := make([]net.Addr, 0, len(addrs))
+	closeAll := func() {
+		for _, c := range conns {
+			_ = c.Close()
+		}
+	}
+	for _, addr := range addrs {
+		udpAddr, err := net.ResolveUDPAddr("udp", net.JoinHostPort(addr, strconv.Itoa(int(port))))
+		if err != nil {
+			closeAll()
+			return nil, nil, err
+		}
+		udp, err := net.ListenUDP("udp", udpAddr)
+		if err != nil {
+			closeAll()
+			return nil, nil, err
+		}
+		conns = append(conns, udp)
+		bound = append(bound, udp.LocalAddr())
+	}
+
+	for _, udp := range conns {
+		go func() {
+			if err := wt.Serve(udp); err != nil {
+				logger.Infow("webtransport listener stopped", "error", err)
+			}
+		}()
+	}
+	logger.Infow("webtransport listener started", "addresses", bound)
+
+	return bound, func() {
+		_ = wt.Close()
+		closeAll()
+	}, nil
+}

pkg/service/webtransport_test.go

@@ -0,0 +1,47 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package service
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/urfave/negroni/v3"
+)
+
+// An upgrade type-asserts the ResponseWriter to http3.Settingser and
+// http3.HTTPStreamer without checking, so it must reach the one net/http handed
+// in. Every negroni chain wraps it.
+func TestUnwrapResponseWriterReachesTheOriginal(t *testing.T) {
+	original := httptest.NewRecorder()
+
+	var seen http.ResponseWriter
+	chain := negroni.New()
+	chain.UseHandler(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		seen = w
+	}))
+	chain.ServeHTTP(original, httptest.NewRequest(http.MethodGet, "/agent", nil))
+
+	_, wrapped := seen.(interface{ Unwrap() http.ResponseWriter })
+	require.True(t, wrapped, "negroni no longer wraps; this test no longer covers anything")
+	require.Same(t, original, unwrapResponseWriter(seen))
+}
+
+func TestUnwrapResponseWriterPassesThroughUnwrapped(t *testing.T) {
+	original := httptest.NewRecorder()
+	require.Same(t, original, unwrapResponseWriter(original))
+}

pkg/service/wire.go

@@ -38,6 +38,7 @@ import (
 	"github.com/livekit/psrpc/pkg/middleware/otelpsrpc"
 
 	"github.com/livekit/livekit-server/pkg/agent"
+	"github.com/livekit/livekit-server/pkg/agent/endpoint"
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/routing"
 	"github.com/livekit/livekit-server/pkg/sfu"
@@ -80,7 +81,11 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
 		NewRoomService,
 		NewRTCService,
 		NewWHIPService,
-		NewAgentService,
+		endpoint.NewRegistry,
+		NewAgentHandler,
+		NewAgentWSService,
+		NewAgentWTService,
+		NewAgentEndpointService,
 		NewAgentDispatchService,
 		getAgentConfig,
 		agent.NewAgentClient,

pkg/service/wire_gen.go

@@ -8,7 +8,15 @@ package service
 
 import (
 	"fmt"
+	"os"
+
+	"github.com/pion/turn/v5"
+	"github.com/pkg/errors"
+	"github.com/redis/go-redis/v9"
+	"gopkg.in/yaml.v3"
+
 	"github.com/livekit/livekit-server/pkg/agent"
+	"github.com/livekit/livekit-server/pkg/agent/endpoint"
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/routing"
 	"github.com/livekit/livekit-server/pkg/sfu"
@@ -22,11 +30,6 @@ import (
 	"github.com/livekit/protocol/webhook"
 	"github.com/livekit/psrpc"
 	"github.com/livekit/psrpc/pkg/middleware/otelpsrpc"
-	"github.com/pion/turn/v5"
-	"github.com/pkg/errors"
-	"github.com/redis/go-redis/v9"
-	"gopkg.in/yaml.v3"
-	"os"
 )
 
 // Injectors from wire.go:
@@ -125,10 +128,14 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
 	if err != nil {
 		return nil, err
 	}
-	agentService, err := NewAgentService(conf, currentNode, v, keyProvider)
+	registry := endpoint.NewRegistry()
+	agentHandler, err := NewAgentHandler(conf, currentNode, v, keyProvider, registry)
 	if err != nil {
 		return nil, err
 	}
+	agentWSService := NewAgentWSService(conf, agentHandler)
+	agentWTService := NewAgentWTService(agentHandler)
+	agentEndpointService := NewAgentEndpointService(agentHandler, registry)
 	agentConfig := getAgentConfig(conf)
 	client, err := agent.NewAgentClient(v, agentConfig)
 	if err != nil {
@@ -151,7 +158,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
 	if err != nil {
 		return nil, err
 	}
-	livekitServer, err := NewLivekitServer(conf, roomService, agentDispatchService, egressService, ingressService, sipService, ioInfoService, rtcService, serviceWHIPService, agentService, keyProvider, router, roomManager, signalServer, server, currentNode)
+	livekitServer, err := NewLivekitServer(conf, roomService, agentDispatchService, egressService, ingressService, sipService, ioInfoService, rtcService, serviceWHIPService, agentWSService, agentWTService, agentEndpointService, keyProvider, router, roomManager, signalServer, server, currentNode)
 	if err != nil {
 		return nil, err
 	}