diff --git a/xds/internal/clients/grpctransport/grpc_transport.go b/xds/internal/clients/grpctransport/grpc_transport.go
new file mode 100644
index 000000000000..c5c1f99694ba
--- /dev/null
+++ b/xds/internal/clients/grpctransport/grpc_transport.go
@@ -0,0 +1,141 @@
+/*
+ *
+ * Copyright 2025 gRPC authors.
+ *
+ * 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 grpctransport provides an implementation of the
+// clients.TransportBuilder interface using gRPC.
+package grpctransport
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/keepalive"
+	"google.golang.org/grpc/xds/internal/clients"
+)
+
+// ServerConfigExtension holds settings for connecting to a gRPC server,
+// such as an xDS management or an LRS server.
+type ServerConfigExtension struct {
+	// Credentials will be used for all gRPC transports. If it is unset,
+	// transport creation will fail.
+	Credentials credentials.Bundle
+}
+
+// Builder creates gRPC-based Transports. It must be paired with ServerConfigs
+// that contain an Extension field of type ServerConfigExtension.
+type Builder struct{}
+
+// Build returns a gRPC-based clients.Transport.
+//
+// The Extension field of the ServerConfig must be a ServerConfigExtension.
+func (b *Builder) Build(sc clients.ServerConfig) (clients.Transport, error) {
+	if sc.ServerURI == "" {
+		return nil, fmt.Errorf("grpctransport: ServerURI is not set in ServerConfig")
+	}
+	if sc.Extensions == nil {
+		return nil, fmt.Errorf("grpctransport: Extensions is not set in ServerConfig")
+	}
+	sce, ok := sc.Extensions.(ServerConfigExtension)
+	if !ok {
+		return nil, fmt.Errorf("grpctransport: Extensions field is %T, but must be %T in ServerConfig", sc.Extensions, ServerConfigExtension{})
+	}
+	if sce.Credentials == nil {
+		return nil, fmt.Errorf("grptransport: Credentials field is not set in ServerConfigExtension")
+	}
+
+	// TODO: Incorporate reference count map for existing transports and
+	// deduplicate transports based on the provided ServerConfig so that
+	// transport channel to same server can be shared between xDS and LRS
+	// client.
+
+	// Create a new gRPC client/channel for the server with the provided
+	// credentials, server URI, and a byte codec to send and receive messages.
+	// Also set a static keepalive configuration that is common across gRPC
+	// language implementations.
+	kpCfg := grpc.WithKeepaliveParams(keepalive.ClientParameters{
+		Time:    5 * time.Minute,
+		Timeout: 20 * time.Second,
+	})
+	cc, err := grpc.NewClient(sc.ServerURI, kpCfg, grpc.WithCredentialsBundle(sce.Credentials), grpc.WithDefaultCallOptions(grpc.ForceCodec(&byteCodec{})))
+	if err != nil {
+		return nil, fmt.Errorf("grpctransport: failed to create transport to server %q: %v", sc.ServerURI, err)
+	}
+
+	return &grpcTransport{cc: cc}, nil
+}
+
+type grpcTransport struct {
+	cc *grpc.ClientConn
+}
+
+// NewStream creates a new gRPC stream to the server for the specified method.
+func (g *grpcTransport) NewStream(ctx context.Context, method string) (clients.Stream, error) {
+	s, err := g.cc.NewStream(ctx, &grpc.StreamDesc{ClientStreams: true, ServerStreams: true}, method)
+	if err != nil {
+		return nil, err
+	}
+	return &stream{stream: s}, nil
+}
+
+// Close closes the gRPC channel to the server.
+func (g *grpcTransport) Close() error {
+	return g.cc.Close()
+}
+
+type stream struct {
+	stream grpc.ClientStream
+}
+
+// Send sends a message to the server.
+func (s *stream) Send(msg []byte) error {
+	return s.stream.SendMsg(msg)
+}
+
+// Recv receives a message from the server.
+func (s *stream) Recv() ([]byte, error) {
+	var typedRes []byte
+
+	if err := s.stream.RecvMsg(&typedRes); err != nil {
+		return nil, err
+	}
+	return typedRes, nil
+}
+
+type byteCodec struct{}
+
+func (c *byteCodec) Marshal(v any) ([]byte, error) {
+	if b, ok := v.([]byte); ok {
+		return b, nil
+	}
+	return nil, fmt.Errorf("grpctransport: message is %T, but must be a []byte", v)
+}
+
+func (c *byteCodec) Unmarshal(data []byte, v any) error {
+	if b, ok := v.(*[]byte); ok {
+		*b = data
+		return nil
+	}
+	return fmt.Errorf("grpctransport: target is %T, but must be *[]byte", v)
+}
+
+func (c *byteCodec) Name() string {
+	return "grpctransport.byteCodec"
+}
diff --git a/xds/internal/clients/grpctransport/grpc_transport_test.go b/xds/internal/clients/grpctransport/grpc_transport_test.go
new file mode 100644
index 000000000000..0ab48707c05e
--- /dev/null
+++ b/xds/internal/clients/grpctransport/grpc_transport_test.go
@@ -0,0 +1,318 @@
+/*
+ *
+ * Copyright 2025 gRPC authors.
+ *
+ * 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 grpctransport
+
+import (
+	"context"
+	"io"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/credentials/local"
+	"google.golang.org/grpc/internal/grpctest"
+	"google.golang.org/grpc/xds/internal/clients"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/testing/protocmp"
+
+	v3discoverygrpc "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
+	v3discoverypb "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
+)
+
+const (
+	defaultTestTimeout = 10 * time.Second
+)
+
+type s struct {
+	grpctest.Tester
+}
+
+func Test(t *testing.T) {
+	grpctest.RunSubTests(t, s{})
+}
+
+// testServer implements the AggregatedDiscoveryServiceServer interface to test
+// the gRPC transport implementation.
+type testServer struct {
+	v3discoverygrpc.UnimplementedAggregatedDiscoveryServiceServer
+
+	address     string                               // address of the server
+	requestChan chan *v3discoverypb.DiscoveryRequest // channel to send the received requests on for verification
+	response    *v3discoverypb.DiscoveryResponse     // response to send back to the client from handler
+}
+
+// setupTestServer set up the gRPC server for AggregatedDiscoveryService. It
+// creates an instance of testServer that returns the provided response from
+// the StreamAggregatedResources() handler and registers it with a gRPC server.
+func setupTestServer(t *testing.T, response *v3discoverypb.DiscoveryResponse) *testServer {
+	t.Helper()
+
+	lis, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatalf("Failed to listen on localhost:0: %v", err)
+	}
+	ts := &testServer{
+		requestChan: make(chan *v3discoverypb.DiscoveryRequest),
+		address:     lis.Addr().String(),
+		response:    response,
+	}
+
+	s := grpc.NewServer()
+
+	v3discoverygrpc.RegisterAggregatedDiscoveryServiceServer(s, ts)
+	go s.Serve(lis)
+	t.Cleanup(s.Stop)
+
+	return ts
+}
+
+// StreamAggregatedResources handles bidirectional streaming of
+// DiscoveryRequest and DiscoveryResponse. It waits for a message from the
+// client on the stream, and then sends a discovery response message back to
+// the client. It also put the received message in requestChan for client to
+// verify if the correct request was received. It continues until the client
+// closes the stream.
+func (s *testServer) StreamAggregatedResources(stream v3discoverygrpc.AggregatedDiscoveryService_StreamAggregatedResourcesServer) error {
+	ctx := stream.Context()
+
+	for {
+		// Receive a DiscoveryRequest from the client
+		req, err := stream.Recv()
+		if err == io.EOF {
+			return nil // Stream closed by client
+		}
+		if err != nil {
+			return err // Handle other errors
+		}
+
+		select {
+		case s.requestChan <- req:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+
+		// Send the response back to the client
+		if err := stream.Send(s.response); err != nil {
+			return err
+		}
+	}
+}
+
+type testCredentials struct {
+	credentials.Bundle
+	transportCredentials credentials.TransportCredentials
+}
+
+func (tc *testCredentials) TransportCredentials() credentials.TransportCredentials {
+	return tc.transportCredentials
+}
+
+// TestBuild_Success verifies that the Builder successfully creates a new
+// Transport with a non-nil grpc.ClientConn.
+func (s) TestBuild_Success(t *testing.T) {
+	serverCfg := clients.ServerConfig{
+		ServerURI:  "server-address",
+		Extensions: ServerConfigExtension{Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
+	}
+
+	b := &Builder{}
+	tr, err := b.Build(serverCfg)
+	if err != nil {
+		t.Fatalf("Build() failed: %v", err)
+	}
+	defer tr.Close()
+
+	if tr == nil {
+		t.Fatalf("Got nil transport from Build(), want non-nil")
+	}
+	if tr.(*grpcTransport).cc == nil {
+		t.Fatalf("Got nil grpc.ClientConn in transport, want non-nil")
+	}
+}
+
+// TestBuild_Failure verifies that the Builder returns error when incorrect
+// ServerConfig is provided.
+//
+// It covers the following scenarios:
+// - ServerURI is empty.
+// - Extensions is nil.
+// - Extensions is not ServerConfigExtension.
+// - Credentials are nil.
+func (s) TestBuild_Failure(t *testing.T) {
+	tests := []struct {
+		name      string
+		serverCfg clients.ServerConfig
+	}{
+		{
+			name: "ServerURI is empty",
+			serverCfg: clients.ServerConfig{
+				ServerURI:  "",
+				Extensions: ServerConfigExtension{Credentials: insecure.NewBundle()},
+			},
+		},
+		{
+			name:      "Extensions is nil",
+			serverCfg: clients.ServerConfig{ServerURI: "server-address"},
+		},
+		{
+			name: "Extensions is not a ServerConfigExtension",
+			serverCfg: clients.ServerConfig{
+				ServerURI:  "server-address",
+				Extensions: 1,
+			},
+		},
+		{
+			name: "ServerConfigExtension Credentials is nil",
+			serverCfg: clients.ServerConfig{
+				ServerURI:  "server-address",
+				Extensions: ServerConfigExtension{},
+			},
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			b := &Builder{}
+			tr, err := b.Build(test.serverCfg)
+			if err == nil {
+				t.Fatalf("Build() succeeded, want error")
+			}
+			if tr != nil {
+				t.Fatalf("Got non-nil transport from Build(), want nil")
+			}
+		})
+	}
+}
+
+// TestNewStream_Success verifies that NewStream() successfully creates a new
+// client stream for the server when provided a valid server URI.
+func (s) TestNewStream_Success(t *testing.T) {
+	ts := setupTestServer(t, &v3discoverypb.DiscoveryResponse{VersionInfo: "1"})
+
+	serverCfg := clients.ServerConfig{
+		ServerURI:  ts.address,
+		Extensions: ServerConfigExtension{Credentials: insecure.NewBundle()},
+	}
+	builder := Builder{}
+	transport, err := builder.Build(serverCfg)
+	if err != nil {
+		t.Fatalf("Failed to build transport: %v", err)
+	}
+	defer transport.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	if _, err = transport.NewStream(ctx, "/envoy.service.discovery.v3.AggregatedDiscoveryService/StreamAggregatedResources"); err != nil {
+		t.Fatalf("transport.NewStream() failed: %v", err)
+	}
+}
+
+// TestNewStream_Error verifies that NewStream() returns an error
+// when attempting to create a stream with an invalid server URI.
+func (s) TestNewStream_Error(t *testing.T) {
+	serverCfg := clients.ServerConfig{
+		ServerURI:  "invalid-server-uri",
+		Extensions: ServerConfigExtension{Credentials: insecure.NewBundle()},
+	}
+	builder := Builder{}
+	transport, err := builder.Build(serverCfg)
+	if err != nil {
+		t.Fatalf("Failed to build transport: %v", err)
+	}
+	defer transport.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	if _, err = transport.NewStream(ctx, "/envoy.service.discovery.v3.AggregatedDiscoveryService/StreamAggregatedResources"); err == nil {
+		t.Fatal("transport.NewStream() succeeded, want failure")
+	}
+}
+
+// TestStream_SendAndRecv verifies that Send() and Recv() successfully send
+// and receive messages on the stream to and from the gRPC server.
+//
+// It starts a gRPC test server using setupTestServer(). The test then sends a
+// testDiscoverRequest on the stream and verifies that the received discovery
+// request on the server is same as sent. It then wait to receive a
+// testDiscoverResponse from the server and verifies that the received
+// discovery response is same as sent from the server.
+func (s) TestStream_SendAndRecv(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout*2000)
+	defer cancel()
+
+	ts := setupTestServer(t, &v3discoverypb.DiscoveryResponse{VersionInfo: "1"})
+
+	// Build a grpc-based transport to the above server.
+	serverCfg := clients.ServerConfig{
+		ServerURI:  ts.address,
+		Extensions: ServerConfigExtension{Credentials: insecure.NewBundle()},
+	}
+	builder := Builder{}
+	transport, err := builder.Build(serverCfg)
+	if err != nil {
+		t.Fatalf("Failed to build transport: %v", err)
+	}
+	defer transport.Close()
+
+	// Create a new stream to the server.
+	stream, err := transport.NewStream(ctx, "/envoy.service.discovery.v3.AggregatedDiscoveryService/StreamAggregatedResources")
+	if err != nil {
+		t.Fatalf("Failed to create stream: %v", err)
+	}
+
+	// Send a discovery request message on the stream.
+	testDiscoverRequest := &v3discoverypb.DiscoveryRequest{VersionInfo: "1"}
+	msg, err := proto.Marshal(testDiscoverRequest)
+	if err != nil {
+		t.Fatalf("Failed to marshal DiscoveryRequest: %v", err)
+	}
+	if err := stream.Send(msg); err != nil {
+		t.Fatalf("Failed to send message: %v", err)
+	}
+
+	// Verify that the DiscoveryRequest received on the server was same as
+	// sent.
+	select {
+	case gotReq := <-ts.requestChan:
+		if diff := cmp.Diff(testDiscoverRequest, gotReq, protocmp.Transform()); diff != "" {
+			t.Fatalf("Unexpected diff in request received on server (-want +got):\n%s", diff)
+		}
+	case <-ctx.Done():
+		t.Fatalf("Timeout waiting for request to reach server")
+	}
+
+	// Wait until response message is received from the server.
+	res, err := stream.Recv()
+	if err != nil {
+		t.Fatalf("Failed to receive message: %v", err)
+	}
+
+	// Verify that the DiscoveryResponse received was same as sent from the
+	// server.
+	var gotRes v3discoverypb.DiscoveryResponse
+	if err := proto.Unmarshal(res, &gotRes); err != nil {
+		t.Fatalf("Failed to unmarshal response from server to DiscoveryResponse: %v", err)
+	}
+	if diff := cmp.Diff(ts.response, &gotRes, protocmp.Transform()); diff != "" {
+		t.Fatalf("proto.Unmarshal(res, &gotRes) returned unexpected diff (-want +got):\n%s", diff)
+	}
+}