// Copyright 2025 Google LLC // // 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 postgresreplicationstats import ( "context" "fmt" "net/http" yaml "github.com/goccy/go-yaml" "github.com/googleapis/mcp-toolbox/internal/tools" "github.com/googleapis/mcp-toolbox/internal/util" "github.com/googleapis/mcp-toolbox/internal/util/parameters" "github.com/jackc/pgx/v5/pgxpool" ) const resourceType string = "postgres-replication-stats" const replicationStats = ` SELECT pid, usename, application_name, backend_xmin, client_addr, state, sync_state, pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), sent_lsn)) AS sent_lag, pg_size_pretty(pg_wal_lsn_diff(sent_lsn, write_lsn)) AS write_lag, pg_size_pretty(pg_wal_lsn_diff(write_lsn, flush_lsn)) AS flush_lag, pg_size_pretty(pg_wal_lsn_diff(flush_lsn, replay_lsn)) AS replay_lag, pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn)) AS total_lag FROM pg_stat_replication; ` func init() { if !tools.Register(resourceType, newConfig) { panic(fmt.Sprintf("tool type %q already registered", resourceType)) } } func newConfig(ctx context.Context, name string, decoder *yaml.Decoder) (tools.ToolConfig, error) { actual := Config{ConfigBase: tools.ConfigBase{Name: name}} if err := decoder.DecodeContext(ctx, &actual); err != nil { return nil, err } return actual, nil } type compatibleSource interface { PostgresPool() *pgxpool.Pool RunSQL(context.Context, string, []any) (any, error) } type Config struct { tools.ConfigBase `yaml:",inline"` Type string `yaml:"type" validate:"required"` Source string `yaml:"source" validate:"required"` Annotations *tools.ToolAnnotations `yaml:"annotations,omitempty"` } var _ tools.ToolConfig = Config{} func (cfg Config) ToolConfigType() string { return resourceType } func (cfg Config) Initialize(context.Context) (tools.Tool, error) { allParameters := parameters.Parameters{} if cfg.Description == "" { cfg.Description = "Lists each replica's process ID, user name, application name, backend_xmin (standby's xmin horizon reported by hot_standby_feedback), client IP address, connection state, and sync_state, along with lag sizes in bytes for sent_lag (primary to sent), write_lag (sent to written), flush_lag (written to flushed), replay_lag (flushed to replayed), and the overall total_lag (primary to replayed)." } return Tool{ BaseTool: tools.NewBaseTool( cfg, tools.GetAnnotationsOrDefault(cfg.Annotations, tools.NewReadOnlyAnnotations), tools.Manifest{Description: cfg.Description, Parameters: allParameters.Manifest(), AuthRequired: cfg.AuthRequired}, allParameters, ), }, nil } var _ tools.Tool = Tool{} type Tool struct { tools.BaseTool[Config] } func (t Tool) Invoke(ctx context.Context, resourceMgr tools.SourceProvider, params parameters.ParamValues, accessToken tools.AccessToken) (any, util.ToolboxError) { source, err := tools.GetCompatibleSource[compatibleSource](resourceMgr, t.Cfg.Source, t.Cfg.Name, t.Cfg.Type) if err != nil { return nil, util.NewClientServerError("source used is not compatible with the tool", http.StatusInternalServerError, err) } paramsMap := params.AsMap() newParams, err := parameters.GetParams(t.StaticParameters, paramsMap) if err != nil { return nil, util.NewAgentError("unable to extract standard params", err) } sliceParams := newParams.AsSlice() resp, err := source.RunSQL(ctx, replicationStats, sliceParams) if err != nil { return nil, util.ProcessGeneralError(err) } return resp, nil } func (t Tool) ToConfig() tools.ToolConfig { return t.Cfg }