refactor(sources/cloudsqladmin): move source implementation in Invoke() function to Source (#2233)

Move source-related queries from `Invoke()` function into Source.

This is an effort to generalizing tools to work with any Source that
implements a specific interface. This will provide a better segregation
of the roles for Tools vs Source.

Tool's role will be limited to the following:
* Resolve any pre-implementation steps or parameters (e.g. template
parameters)
* Retrieving Source
* Calling the source's implementation


Along with these updates, this PR also resolve some comments from
Gemini:
* update `fmt.Printf()` to logging as a Debug log and remove the
training `\n` within the log
* move `regexp.MustCompile` to the top so that it's compiled once at the
package level and reused. It is a relatively expensive operation to be
called on every invocation.
* `fetchInstanceData()` to return the `*sqladmin.DatabaseInstance`
struct directly instead of converting to map and use map lookups. More
typesafe and efficient.


Did not move `cloudsqlpgupgradeprecheck` tool since that invocation is
very specific towards cloudsql for postgres

internal/sources/cloudsqladmin/cloud_sql_admin.go

@@ -15,10 +15,16 @@ package cloudsqladmin
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net/http"
+	"regexp"
+	"strings"
+	"text/template"
+	"time"
 
 	"github.com/goccy/go-yaml"
+	"github.com/googleapis/genai-toolbox/internal/log"
 	"github.com/googleapis/genai-toolbox/internal/sources"
 	"github.com/googleapis/genai-toolbox/internal/util"
 	"go.opentelemetry.io/otel/trace"
@@ -30,6 +36,8 @@ import (
 
 const SourceKind string = "cloud-sql-admin"
 
+var targetLinkRegex = regexp.MustCompile(`/projects/([^/]+)/instances/([^/]+)/databases/([^/]+)`)
+
 // validate interface
 var _ sources.SourceConfig = Config{}
 
@@ -130,3 +138,304 @@ func (s *Source) GetService(ctx context.Context, accessToken string) (*sqladmin.
 func (s *Source) UseClientAuthorization() bool {
 	return s.UseClientOAuth
 }
+
+func (s *Source) CloneInstance(ctx context.Context, project, sourceInstanceName, destinationInstanceName, pointInTime, preferredZone, preferredSecondaryZone, accessToken string) (any, error) {
+	cloneContext := &sqladmin.CloneContext{
+		DestinationInstanceName: destinationInstanceName,
+	}
+
+	if pointInTime != "" {
+		cloneContext.PointInTime = pointInTime
+	}
+	if preferredZone != "" {
+		cloneContext.PreferredZone = preferredZone
+	}
+	if preferredSecondaryZone != "" {
+		cloneContext.PreferredSecondaryZone = preferredSecondaryZone
+	}
+
+	rb := &sqladmin.InstancesCloneRequest{
+		CloneContext: cloneContext,
+	}
+	service, err := s.GetService(ctx, accessToken)
+	if err != nil {
+		return nil, err
+	}
+	resp, err := service.Instances.Clone(project, sourceInstanceName, rb).Do()
+	if err != nil {
+		return nil, fmt.Errorf("error cloning instance: %w", err)
+	}
+	return resp, nil
+}
+
+func (s *Source) CreateDatabase(ctx context.Context, name, project, instance, accessToken string) (any, error) {
+	database := sqladmin.Database{
+		Name:     name,
+		Project:  project,
+		Instance: instance,
+	}
+
+	service, err := s.GetService(ctx, accessToken)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := service.Databases.Insert(project, instance, &database).Do()
+	if err != nil {
+		return nil, fmt.Errorf("error creating database: %w", err)
+	}
+	return resp, nil
+}
+
+func (s *Source) CreateUsers(ctx context.Context, project, instance, name, password string, iamUser bool, accessToken string) (any, error) {
+	service, err := s.GetService(ctx, accessToken)
+	if err != nil {
+		return nil, err
+	}
+
+	user := sqladmin.User{
+		Name: name,
+	}
+
+	if iamUser {
+		user.Type = "CLOUD_IAM_USER"
+	} else {
+		user.Type = "BUILT_IN"
+		if password == "" {
+			return nil, fmt.Errorf("missing 'password' parameter for non-IAM user")
+		}
+		user.Password = password
+	}
+
+	resp, err := service.Users.Insert(project, instance, &user).Do()
+	if err != nil {
+		return nil, fmt.Errorf("error creating user: %w", err)
+	}
+
+	return resp, nil
+}
+
+func (s *Source) GetInstance(ctx context.Context, projectId, instanceId, accessToken string) (any, error) {
+	service, err := s.GetService(ctx, accessToken)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := service.Instances.Get(projectId, instanceId).Do()
+	if err != nil {
+		return nil, fmt.Errorf("error getting instance: %w", err)
+	}
+	return resp, nil
+}
+
+func (s *Source) ListDatabase(ctx context.Context, project, instance, accessToken string) (any, error) {
+	service, err := s.GetService(ctx, accessToken)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := service.Databases.List(project, instance).Do()
+	if err != nil {
+		return nil, fmt.Errorf("error listing databases: %w", err)
+	}
+
+	if resp.Items == nil {
+		return []any{}, nil
+	}
+
+	type databaseInfo struct {
+		Name      string `json:"name"`
+		Charset   string `json:"charset"`
+		Collation string `json:"collation"`
+	}
+
+	var databases []databaseInfo
+	for _, item := range resp.Items {
+		databases = append(databases, databaseInfo{
+			Name:      item.Name,
+			Charset:   item.Charset,
+			Collation: item.Collation,
+		})
+	}
+	return databases, nil
+}
+
+func (s *Source) ListInstance(ctx context.Context, project, accessToken string) (any, error) {
+	service, err := s.GetService(ctx, accessToken)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := service.Instances.List(project).Do()
+	if err != nil {
+		return nil, fmt.Errorf("error listing instances: %w", err)
+	}
+
+	if resp.Items == nil {
+		return []any{}, nil
+	}
+
+	type instanceInfo struct {
+		Name         string `json:"name"`
+		InstanceType string `json:"instanceType"`
+	}
+
+	var instances []instanceInfo
+	for _, item := range resp.Items {
+		instances = append(instances, instanceInfo{
+			Name:         item.Name,
+			InstanceType: item.InstanceType,
+		})
+	}
+	return instances, nil
+}
+
+func (s *Source) CreateInstance(ctx context.Context, project, name, dbVersion, rootPassword string, settings sqladmin.Settings, accessToken string) (any, error) {
+	instance := sqladmin.DatabaseInstance{
+		Name:            name,
+		DatabaseVersion: dbVersion,
+		RootPassword:    rootPassword,
+		Settings:        &settings,
+		Project:         project,
+	}
+
+	service, err := s.GetService(ctx, accessToken)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := service.Instances.Insert(project, &instance).Do()
+	if err != nil {
+		return nil, fmt.Errorf("error creating instance: %w", err)
+	}
+
+	return resp, nil
+}
+
+func (s *Source) GetWaitForOperations(ctx context.Context, service *sqladmin.Service, project, operation, connectionMessageTemplate string, delay time.Duration) (any, error) {
+	logger, err := util.LoggerFromContext(ctx)
+	if err != nil {
+		return nil, err
+	}
+	op, err := service.Operations.Get(project, operation).Do()
+	if err != nil {
+		logger.DebugContext(ctx, fmt.Sprintf("error getting operation: %s, retrying in %v", err, delay))
+	} else {
+		if op.Status == "DONE" {
+			if op.Error != nil {
+				var errorBytes []byte
+				errorBytes, err = json.Marshal(op.Error)
+				if err != nil {
+					return nil, fmt.Errorf("operation finished with error but could not marshal error object: %w", err)
+				}
+				return nil, fmt.Errorf("operation finished with error: %s", string(errorBytes))
+			}
+
+			var opBytes []byte
+			opBytes, err = op.MarshalJSON()
+			if err != nil {
+				return nil, fmt.Errorf("could not marshal operation: %w", err)
+			}
+
+			var data map[string]any
+			if err := json.Unmarshal(opBytes, &data); err != nil {
+				return nil, fmt.Errorf("could not unmarshal operation: %w", err)
+			}
+
+			if msg, ok := generateCloudSQLConnectionMessage(ctx, s, logger, data, connectionMessageTemplate); ok {
+				return msg, nil
+			}
+			return string(opBytes), nil
+		}
+		logger.DebugContext(ctx, fmt.Sprintf("operation not complete, retrying in %v", delay))
+	}
+	return nil, nil
+}
+
+func generateCloudSQLConnectionMessage(ctx context.Context, source *Source, logger log.Logger, opResponse map[string]any, connectionMessageTemplate string) (string, bool) {
+	operationType, ok := opResponse["operationType"].(string)
+	if !ok || operationType != "CREATE_DATABASE" {
+		return "", false
+	}
+
+	targetLink, ok := opResponse["targetLink"].(string)
+	if !ok {
+		return "", false
+	}
+
+	matches := targetLinkRegex.FindStringSubmatch(targetLink)
+	if len(matches) < 4 {
+		return "", false
+	}
+	project := matches[1]
+	instance := matches[2]
+	database := matches[3]
+
+	dbInstance, err := fetchInstanceData(ctx, source, project, instance)
+	if err != nil {
+		logger.DebugContext(ctx, fmt.Sprintf("error fetching instance data: %v", err))
+		return "", false
+	}
+
+	region := dbInstance.Region
+	if region == "" {
+		return "", false
+	}
+
+	databaseVersion := dbInstance.DatabaseVersion
+	if databaseVersion == "" {
+		return "", false
+	}
+
+	var dbType string
+	if strings.Contains(databaseVersion, "POSTGRES") {
+		dbType = "postgres"
+	} else if strings.Contains(databaseVersion, "MYSQL") {
+		dbType = "mysql"
+	} else if strings.Contains(databaseVersion, "SQLSERVER") {
+		dbType = "mssql"
+	} else {
+		return "", false
+	}
+
+	tmpl, err := template.New("cloud-sql-connection").Parse(connectionMessageTemplate)
+	if err != nil {
+		return fmt.Sprintf("template parsing error: %v", err), false
+	}
+
+	data := struct {
+		Project     string
+		Region      string
+		Instance    string
+		DBType      string
+		DBTypeUpper string
+		Database    string
+	}{
+		Project:     project,
+		Region:      region,
+		Instance:    instance,
+		DBType:      dbType,
+		DBTypeUpper: strings.ToUpper(dbType),
+		Database:    database,
+	}
+
+	var b strings.Builder
+	if err := tmpl.Execute(&b, data); err != nil {
+		return fmt.Sprintf("template execution error: %v", err), false
+	}
+
+	return b.String(), true
+}
+
+func fetchInstanceData(ctx context.Context, source *Source, project, instance string) (*sqladmin.DatabaseInstance, error) {
+	service, err := source.GetService(ctx, "")
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := service.Instances.Get(project, instance).Do()
+	if err != nil {
+		return nil, fmt.Errorf("error getting instance: %w", err)
+	}
+	return resp, nil
+}

internal/tools/cloudsql/cloudsqlcloneinstance/cloudsqlcloneinstance.go

@@ -45,6 +45,7 @@ type compatibleSource interface {
 	GetDefaultProject() string
 	GetService(context.Context, string) (*sqladmin.Service, error)
 	UseClientAuthorization() bool
+	CloneInstance(context.Context, string, string, string, string, string, string, string) (any, error)
 }
 
 // Config defines the configuration for the clone-instance tool.
@@ -142,38 +143,11 @@ func (t Tool) Invoke(ctx context.Context, resourceMgr tools.SourceProvider, para
 		return nil, fmt.Errorf("error casting 'destinationInstanceName' parameter: %v", paramsMap["destinationInstanceName"])
 	}
 
-	cloneContext := &sqladmin.CloneContext{
-		DestinationInstanceName: destinationInstanceName,
-	}
+	pointInTime, _ := paramsMap["pointInTime"].(string)
+	preferredZone, _ := paramsMap["preferredZone"].(string)
+	preferredSecondaryZone, _ := paramsMap["preferredSecondaryZone"].(string)
 
-	pointInTime, ok := paramsMap["pointInTime"].(string)
-	if ok {
-		cloneContext.PointInTime = pointInTime
-	}
-	preferredZone, ok := paramsMap["preferredZone"].(string)
-	if ok {
-		cloneContext.PreferredZone = preferredZone
-	}
-	preferredSecondaryZone, ok := paramsMap["preferredSecondaryZone"].(string)
-	if ok {
-		cloneContext.PreferredSecondaryZone = preferredSecondaryZone
-	}
-
-	rb := &sqladmin.InstancesCloneRequest{
-		CloneContext: cloneContext,
-	}
-
-	service, err := source.GetService(ctx, string(accessToken))
-	if err != nil {
-		return nil, err
-	}
-
-	resp, err := service.Instances.Clone(project, sourceInstanceName, rb).Do()
-	if err != nil {
-		return nil, fmt.Errorf("error cloning instance: %w", err)
-	}
-
-	return resp, nil
+	return source.CloneInstance(ctx, project, sourceInstanceName, destinationInstanceName, pointInTime, preferredZone, preferredSecondaryZone, string(accessToken))
 }
 
 // ParseParams parses the parameters for the tool.

internal/tools/cloudsql/cloudsqlcreatedatabase/cloudsqlcreatedatabase.go

@@ -22,7 +22,6 @@ import (
 	"github.com/googleapis/genai-toolbox/internal/sources"
 	"github.com/googleapis/genai-toolbox/internal/tools"
 	"github.com/googleapis/genai-toolbox/internal/util/parameters"
-	sqladmin "google.golang.org/api/sqladmin/v1"
 )
 
 const kind string = "cloud-sql-create-database"
@@ -43,8 +42,8 @@ func newConfig(ctx context.Context, name string, decoder *yaml.Decoder) (tools.T
 
 type compatibleSource interface {
 	GetDefaultProject() string
-	GetService(context.Context, string) (*sqladmin.Service, error)
 	UseClientAuthorization() bool
+	CreateDatabase(context.Context, string, string, string, string) (any, error)
 }
 
 // Config defines the configuration for the create-database tool.
@@ -137,24 +136,7 @@ func (t Tool) Invoke(ctx context.Context, resourceMgr tools.SourceProvider, para
 	if !ok {
 		return nil, fmt.Errorf("missing 'name' parameter")
 	}
-
-	database := sqladmin.Database{
-		Name:     name,
-		Project:  project,
-		Instance: instance,
-	}
-
-	service, err := source.GetService(ctx, string(accessToken))
-	if err != nil {
-		return nil, err
-	}
-
-	resp, err := service.Databases.Insert(project, instance, &database).Do()
-	if err != nil {
-		return nil, fmt.Errorf("error creating database: %w", err)
-	}
-
-	return resp, nil
+	return source.CreateDatabase(ctx, name, project, instance, string(accessToken))
 }
 
 // ParseParams parses the parameters for the tool.

internal/tools/cloudsql/cloudsqlcreateusers/cloudsqlcreateusers.go

@@ -22,7 +22,6 @@ import (
 	"github.com/googleapis/genai-toolbox/internal/sources"
 	"github.com/googleapis/genai-toolbox/internal/tools"
 	"github.com/googleapis/genai-toolbox/internal/util/parameters"
-	sqladmin "google.golang.org/api/sqladmin/v1"
 )
 
 const kind string = "cloud-sql-create-users"
@@ -43,8 +42,8 @@ func newConfig(ctx context.Context, name string, decoder *yaml.Decoder) (tools.T
 
 type compatibleSource interface {
 	GetDefaultProject() string
-	GetService(context.Context, string) (*sqladmin.Service, error)
 	UseClientAuthorization() bool
+	CreateUsers(context.Context, string, string, string, string, bool, string) (any, error)
 }
 
 // Config defines the configuration for the create-user tool.
@@ -141,33 +140,8 @@ func (t Tool) Invoke(ctx context.Context, resourceMgr tools.SourceProvider, para
 	}
 
 	iamUser, _ := paramsMap["iamUser"].(bool)
-
-	user := sqladmin.User{
-		Name: name,
-	}
-
-	if iamUser {
-		user.Type = "CLOUD_IAM_USER"
-	} else {
-		user.Type = "BUILT_IN"
-		password, ok := paramsMap["password"].(string)
-		if !ok || password == "" {
-			return nil, fmt.Errorf("missing 'password' parameter for non-IAM user")
-		}
-		user.Password = password
-	}
-
-	service, err := source.GetService(ctx, string(accessToken))
-	if err != nil {
-		return nil, err
-	}
-
-	resp, err := service.Users.Insert(project, instance, &user).Do()
-	if err != nil {
-		return nil, fmt.Errorf("error creating user: %w", err)
-	}
-
-	return resp, nil
+	password, _ := paramsMap["password"].(string)
+	return source.CreateUsers(ctx, project, instance, name, password, iamUser, string(accessToken))
 }
 
 // ParseParams parses the parameters for the tool.

internal/tools/cloudsql/cloudsqlgetinstances/cloudsqlgetinstances.go

@@ -22,7 +22,6 @@ import (
 	"github.com/googleapis/genai-toolbox/internal/sources"
 	"github.com/googleapis/genai-toolbox/internal/tools"
 	"github.com/googleapis/genai-toolbox/internal/util/parameters"
-	"google.golang.org/api/sqladmin/v1"
 )
 
 const kind string = "cloud-sql-get-instance"
@@ -43,8 +42,8 @@ func newConfig(ctx context.Context, name string, decoder *yaml.Decoder) (tools.T
 
 type compatibleSource interface {
 	GetDefaultProject() string
-	GetService(context.Context, string) (*sqladmin.Service, error)
 	UseClientAuthorization() bool
+	GetInstance(context.Context, string, string, string) (any, error)
 }
 
 // Config defines the configuration for the get-instances tool.
@@ -133,18 +132,7 @@ func (t Tool) Invoke(ctx context.Context, resourceMgr tools.SourceProvider, para
 	if !ok {
 		return nil, fmt.Errorf("missing 'instanceId' parameter")
 	}
-
-	service, err := source.GetService(ctx, string(accessToken))
-	if err != nil {
-		return nil, err
-	}
-
-	resp, err := service.Instances.Get(projectId, instanceId).Do()
-	if err != nil {
-		return nil, fmt.Errorf("error getting instance: %w", err)
-	}
-
-	return resp, nil
+	return source.GetInstance(ctx, projectId, instanceId, string(accessToken))
 }
 
 // ParseParams parses the parameters for the tool.

internal/tools/cloudsql/cloudsqllistdatabases/cloudsqllistdatabases.go

@@ -22,7 +22,6 @@ import (
 	"github.com/googleapis/genai-toolbox/internal/sources"
 	"github.com/googleapis/genai-toolbox/internal/tools"
 	"github.com/googleapis/genai-toolbox/internal/util/parameters"
-	"google.golang.org/api/sqladmin/v1"
 )
 
 const kind string = "cloud-sql-list-databases"
@@ -43,8 +42,8 @@ func newConfig(ctx context.Context, name string, decoder *yaml.Decoder) (tools.T
 
 type compatibleSource interface {
 	GetDefaultProject() string
-	GetService(context.Context, string) (*sqladmin.Service, error)
 	UseClientAuthorization() bool
+	ListDatabase(context.Context, string, string, string) (any, error)
 }
 
 // Config defines the configuration for the list-databases tool.
@@ -132,37 +131,7 @@ func (t Tool) Invoke(ctx context.Context, resourceMgr tools.SourceProvider, para
 	if !ok {
 		return nil, fmt.Errorf("missing 'instance' parameter")
 	}
-
-	service, err := source.GetService(ctx, string(accessToken))
-	if err != nil {
-		return nil, err
-	}
-
-	resp, err := service.Databases.List(project, instance).Do()
-	if err != nil {
-		return nil, fmt.Errorf("error listing databases: %w", err)
-	}
-
-	if resp.Items == nil {
-		return []any{}, nil
-	}
-
-	type databaseInfo struct {
-		Name      string `json:"name"`
-		Charset   string `json:"charset"`
-		Collation string `json:"collation"`
-	}
-
-	var databases []databaseInfo
-	for _, item := range resp.Items {
-		databases = append(databases, databaseInfo{
-			Name:      item.Name,
-			Charset:   item.Charset,
-			Collation: item.Collation,
-		})
-	}
-
-	return databases, nil
+	return source.ListDatabase(ctx, project, instance, string(accessToken))
 }
 
 // ParseParams parses the parameters for the tool.

internal/tools/cloudsql/cloudsqllistinstances/cloudsqllistinstances.go

@@ -22,7 +22,6 @@ import (
 	"github.com/googleapis/genai-toolbox/internal/sources"
 	"github.com/googleapis/genai-toolbox/internal/tools"
 	"github.com/googleapis/genai-toolbox/internal/util/parameters"
-	"google.golang.org/api/sqladmin/v1"
 )
 
 const kind string = "cloud-sql-list-instances"
@@ -43,8 +42,8 @@ func newConfig(ctx context.Context, name string, decoder *yaml.Decoder) (tools.T
 
 type compatibleSource interface {
 	GetDefaultProject() string
-	GetService(context.Context, string) (*sqladmin.Service, error)
 	UseClientAuthorization() bool
+	ListInstance(context.Context, string, string) (any, error)
 }
 
 // Config defines the configuration for the list-instance tool.
@@ -127,35 +126,7 @@ func (t Tool) Invoke(ctx context.Context, resourceMgr tools.SourceProvider, para
 	if !ok {
 		return nil, fmt.Errorf("missing 'project' parameter")
 	}
-
-	service, err := source.GetService(ctx, string(accessToken))
-	if err != nil {
-		return nil, err
-	}
-
-	resp, err := service.Instances.List(project).Do()
-	if err != nil {
-		return nil, fmt.Errorf("error listing instances: %w", err)
-	}
-
-	if resp.Items == nil {
-		return []any{}, nil
-	}
-
-	type instanceInfo struct {
-		Name         string `json:"name"`
-		InstanceType string `json:"instanceType"`
-	}
-
-	var instances []instanceInfo
-	for _, item := range resp.Items {
-		instances = append(instances, instanceInfo{
-			Name:         item.Name,
-			InstanceType: item.InstanceType,
-		})
-	}
-
-	return instances, nil
+	return source.ListInstance(ctx, project, string(accessToken))
 }
 
 // ParseParams parses the parameters for the tool.

internal/tools/cloudsql/cloudsqlwaitforoperation/cloudsqlwaitforoperation.go

@@ -16,11 +16,7 @@ package cloudsqlwaitforoperation
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
-	"regexp"
-	"strings"
-	"text/template"
 	"time"
 
 	yaml "github.com/goccy/go-yaml"
@@ -91,6 +87,7 @@ type compatibleSource interface {
 	GetDefaultProject() string
 	GetService(context.Context, string) (*sqladmin.Service, error)
 	UseClientAuthorization() bool
+	GetWaitForOperations(context.Context, *sqladmin.Service, string, string, string, time.Duration) (any, error)
 }
 
 // Config defines the configuration for the wait-for-operation tool.
@@ -229,14 +226,14 @@ func (t Tool) Invoke(ctx context.Context, resourceMgr tools.SourceProvider, para
 		return nil, fmt.Errorf("missing 'operation' parameter")
 	}
 
+	ctx, cancel := context.WithTimeout(ctx, 30*time.Minute)
+	defer cancel()
+
 	service, err := source.GetService(ctx, string(accessToken))
 	if err != nil {
 		return nil, err
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, 30*time.Minute)
-	defer cancel()
-
 	delay := t.Delay
 	maxDelay := t.MaxDelay
 	multiplier := t.Multiplier
@@ -250,37 +247,11 @@ func (t Tool) Invoke(ctx context.Context, resourceMgr tools.SourceProvider, para
 		default:
 		}
 
-		op, err := service.Operations.Get(project, operationID).Do()
+		op, err := source.GetWaitForOperations(ctx, service, project, operationID, cloudSQLConnectionMessageTemplate, delay)
 		if err != nil {
-			fmt.Printf("error getting operation: %s, retrying in %v\n", err, delay)
-		} else {
-			if op.Status == "DONE" {
-				if op.Error != nil {
-					var errorBytes []byte
-					errorBytes, err = json.Marshal(op.Error)
-					if err != nil {
-						return nil, fmt.Errorf("operation finished with error but could not marshal error object: %w", err)
-					}
-					return nil, fmt.Errorf("operation finished with error: %s", string(errorBytes))
-				}
-
-				var opBytes []byte
-				opBytes, err = op.MarshalJSON()
-				if err != nil {
-					return nil, fmt.Errorf("could not marshal operation: %w", err)
-				}
-
-				var data map[string]any
-				if err := json.Unmarshal(opBytes, &data); err != nil {
-					return nil, fmt.Errorf("could not unmarshal operation: %w", err)
-				}
-
-				if msg, ok := t.generateCloudSQLConnectionMessage(source, data); ok {
-					return msg, nil
-				}
-				return string(opBytes), nil
-			}
-			fmt.Printf("Operation not complete, retrying in %v\n", delay)
+			return nil, err
+		} else if op != nil {
+			return op, nil
 		}
 
 		time.Sleep(delay)
@@ -321,105 +292,6 @@ func (t Tool) RequiresClientAuthorization(resourceMgr tools.SourceProvider) (boo
 	return source.UseClientAuthorization(), nil
 }
 
-func (t Tool) generateCloudSQLConnectionMessage(source compatibleSource, opResponse map[string]any) (string, bool) {
-	operationType, ok := opResponse["operationType"].(string)
-	if !ok || operationType != "CREATE_DATABASE" {
-		return "", false
-	}
-
-	targetLink, ok := opResponse["targetLink"].(string)
-	if !ok {
-		return "", false
-	}
-
-	r := regexp.MustCompile(`/projects/([^/]+)/instances/([^/]+)/databases/([^/]+)`)
-	matches := r.FindStringSubmatch(targetLink)
-	if len(matches) < 4 {
-		return "", false
-	}
-	project := matches[1]
-	instance := matches[2]
-	database := matches[3]
-
-	instanceData, err := t.fetchInstanceData(context.Background(), source, project, instance)
-	if err != nil {
-		fmt.Printf("error fetching instance data: %v\n", err)
-		return "", false
-	}
-
-	region, ok := instanceData["region"].(string)
-	if !ok {
-		return "", false
-	}
-
-	databaseVersion, ok := instanceData["databaseVersion"].(string)
-	if !ok {
-		return "", false
-	}
-
-	var dbType string
-	if strings.Contains(databaseVersion, "POSTGRES") {
-		dbType = "postgres"
-	} else if strings.Contains(databaseVersion, "MYSQL") {
-		dbType = "mysql"
-	} else if strings.Contains(databaseVersion, "SQLSERVER") {
-		dbType = "mssql"
-	} else {
-		return "", false
-	}
-
-	tmpl, err := template.New("cloud-sql-connection").Parse(cloudSQLConnectionMessageTemplate)
-	if err != nil {
-		return fmt.Sprintf("template parsing error: %v", err), false
-	}
-
-	data := struct {
-		Project     string
-		Region      string
-		Instance    string
-		DBType      string
-		DBTypeUpper string
-		Database    string
-	}{
-		Project:     project,
-		Region:      region,
-		Instance:    instance,
-		DBType:      dbType,
-		DBTypeUpper: strings.ToUpper(dbType),
-		Database:    database,
-	}
-
-	var b strings.Builder
-	if err := tmpl.Execute(&b, data); err != nil {
-		return fmt.Sprintf("template execution error: %v", err), false
-	}
-
-	return b.String(), true
-}
-
-func (t Tool) fetchInstanceData(ctx context.Context, source compatibleSource, project, instance string) (map[string]any, error) {
-	service, err := source.GetService(ctx, "")
-	if err != nil {
-		return nil, err
-	}
-
-	resp, err := service.Instances.Get(project, instance).Do()
-	if err != nil {
-		return nil, fmt.Errorf("error getting instance: %w", err)
-	}
-
-	var data map[string]any
-	var b []byte
-	b, err = resp.MarshalJSON()
-	if err != nil {
-		return nil, fmt.Errorf("error marshalling response: %w", err)
-	}
-	if err := json.Unmarshal(b, &data); err != nil {
-		return nil, fmt.Errorf("error unmarshalling response body: %w", err)
-	}
-	return data, nil
-}
-
 func (t Tool) GetAuthTokenHeaderName(resourceMgr tools.SourceProvider) (string, error) {
 	return "Authorization", nil
 }

internal/tools/cloudsqlmssql/cloudsqlmssqlcreateinstance/cloudsqlmssqlcreateinstance.go

@@ -23,7 +23,7 @@ import (
 	"github.com/googleapis/genai-toolbox/internal/sources"
 	"github.com/googleapis/genai-toolbox/internal/tools"
 	"github.com/googleapis/genai-toolbox/internal/util/parameters"
-	sqladmin "google.golang.org/api/sqladmin/v1"
+	"google.golang.org/api/sqladmin/v1"
 )
 
 const kind string = "cloud-sql-mssql-create-instance"
@@ -44,8 +44,8 @@ func newConfig(ctx context.Context, name string, decoder *yaml.Decoder) (tools.T
 
 type compatibleSource interface {
 	GetDefaultProject() string
-	GetService(context.Context, string) (*sqladmin.Service, error)
 	UseClientAuthorization() bool
+	CreateInstance(context.Context, string, string, string, string, sqladmin.Settings, string) (any, error)
 }
 
 // Config defines the configuration for the create-instances tool.
@@ -148,7 +148,6 @@ func (t Tool) Invoke(ctx context.Context, resourceMgr tools.SourceProvider, para
 	if !ok {
 		return nil, fmt.Errorf("error casting 'editionPreset' parameter: %s", paramsMap["editionPreset"])
 	}
-
 	settings := sqladmin.Settings{}
 	switch strings.ToLower(editionPreset) {
 	case "production":
@@ -166,26 +165,7 @@ func (t Tool) Invoke(ctx context.Context, resourceMgr tools.SourceProvider, para
 	default:
 		return nil, fmt.Errorf("invalid 'editionPreset': %q. Must be either 'Production' or 'Development'", editionPreset)
 	}
-
-	instance := sqladmin.DatabaseInstance{
-		Name:            name,
-		DatabaseVersion: dbVersion,
-		RootPassword:    rootPassword,
-		Settings:        &settings,
-		Project:         project,
-	}
-
-	service, err := source.GetService(ctx, string(accessToken))
-	if err != nil {
-		return nil, err
-	}
-
-	resp, err := service.Instances.Insert(project, &instance).Do()
-	if err != nil {
-		return nil, fmt.Errorf("error creating instance: %w", err)
-	}
-
-	return resp, nil
+	return source.CreateInstance(ctx, project, name, dbVersion, rootPassword, settings, string(accessToken))
 }
 
 // ParseParams parses the parameters for the tool.

internal/tools/cloudsqlmysql/cloudsqlmysqlcreateinstance/cloudsqlmysqlcreateinstance.go

@@ -44,8 +44,8 @@ func newConfig(ctx context.Context, name string, decoder *yaml.Decoder) (tools.T
 
 type compatibleSource interface {
 	GetDefaultProject() string
-	GetService(context.Context, string) (*sqladmin.Service, error)
 	UseClientAuthorization() bool
+	CreateInstance(context.Context, string, string, string, string, sqladmin.Settings, string) (any, error)
 }
 
 // Config defines the configuration for the create-instances tool.
@@ -167,25 +167,7 @@ func (t Tool) Invoke(ctx context.Context, resourceMgr tools.SourceProvider, para
 		return nil, fmt.Errorf("invalid 'editionPreset': %q. Must be either 'Production' or 'Development'", editionPreset)
 	}
 
-	instance := sqladmin.DatabaseInstance{
-		Name:            name,
-		DatabaseVersion: dbVersion,
-		RootPassword:    rootPassword,
-		Settings:        &settings,
-		Project:         project,
-	}
-
-	service, err := source.GetService(ctx, string(accessToken))
-	if err != nil {
-		return nil, err
-	}
-
-	resp, err := service.Instances.Insert(project, &instance).Do()
-	if err != nil {
-		return nil, fmt.Errorf("error creating instance: %w", err)
-	}
-
-	return resp, nil
+	return source.CreateInstance(ctx, project, name, dbVersion, rootPassword, settings, string(accessToken))
 }
 
 // ParseParams parses the parameters for the tool.

internal/tools/cloudsqlpg/cloudsqlpgcreateinstances/cloudsqlpgcreateinstances.go

@@ -44,8 +44,8 @@ func newConfig(ctx context.Context, name string, decoder *yaml.Decoder) (tools.T
 
 type compatibleSource interface {
 	GetDefaultProject() string
-	GetService(context.Context, string) (*sqladmin.Service, error)
 	UseClientAuthorization() bool
+	CreateInstance(context.Context, string, string, string, string, sqladmin.Settings, string) (any, error)
 }
 
 // Config defines the configuration for the create-instances tool.
@@ -166,26 +166,7 @@ func (t Tool) Invoke(ctx context.Context, resourceMgr tools.SourceProvider, para
 	default:
 		return nil, fmt.Errorf("invalid 'editionPreset': %q. Must be either 'Production' or 'Development'", editionPreset)
 	}
-
-	instance := sqladmin.DatabaseInstance{
-		Name:            name,
-		DatabaseVersion: dbVersion,
-		RootPassword:    rootPassword,
-		Settings:        &settings,
-		Project:         project,
-	}
-
-	service, err := source.GetService(ctx, string(accessToken))
-	if err != nil {
-		return nil, err
-	}
-
-	resp, err := service.Instances.Insert(project, &instance).Do()
-	if err != nil {
-		return nil, fmt.Errorf("error creating instance: %w", err)
-	}
-
-	return resp, nil
+	return source.CreateInstance(ctx, project, name, dbVersion, rootPassword, settings, string(accessToken))
 }
 
 // ParseParams parses the parameters for the tool.