gitea-mcp/operation/repo/deploy_key.go

package repo

import (
	"context"
	"fmt"

	"gitea.com/gitea/gitea-mcp/pkg/gitea"
	"gitea.com/gitea/gitea-mcp/pkg/log"
	"gitea.com/gitea/gitea-mcp/pkg/params"
	"gitea.com/gitea/gitea-mcp/pkg/to"

	gitea_sdk "code.gitea.io/sdk/gitea"
	"github.com/mark3labs/mcp-go/mcp"
	"github.com/mark3labs/mcp-go/server"
)

const (
	DeployKeyReadToolName  = "deploy_key_read"
	DeployKeyWriteToolName = "deploy_key_write"
)

var (
	DeployKeyReadTool = mcp.NewTool(
		DeployKeyReadToolName,
		mcp.WithDescription("Read deploy keys. Use method 'list' to list all, 'get' to get a specific key."),
		mcp.WithString("method", mcp.Required(), mcp.Description("operation to perform"), mcp.Enum("list", "get")),
		mcp.WithString("owner", mcp.Required(), mcp.Description("repository owner")),
		mcp.WithString("repo", mcp.Required(), mcp.Description("repository name")),
		mcp.WithNumber("id", mcp.Description("deploy key ID (required for 'get')")),
		mcp.WithNumber("page", mcp.Description("page number"), mcp.DefaultNumber(1)),
		mcp.WithNumber("perPage", mcp.Description("results per page"), mcp.DefaultNumber(30)),
	)

	DeployKeyWriteTool = mcp.NewTool(
		DeployKeyWriteToolName,
		mcp.WithDescription("Create or delete deploy keys. Use method 'create' to add a new key, 'delete' to remove."),
		mcp.WithString("method", mcp.Required(), mcp.Description("operation to perform"), mcp.Enum("create", "delete")),
		mcp.WithString("owner", mcp.Required(), mcp.Description("repository owner")),
		mcp.WithString("repo", mcp.Required(), mcp.Description("repository name")),
		mcp.WithNumber("id", mcp.Description("deploy key ID (required for 'delete')")),
		mcp.WithString("title", mcp.Description("key title (required for 'create')")),
		mcp.WithString("key", mcp.Description("SSH public key content (required for 'create')")),
		mcp.WithBoolean("read_only", mcp.Description("whether the key has read-only access (default true)")),
	)
)

func init() {
	Tool.RegisterRead(server.ServerTool{
		Tool:    DeployKeyReadTool,
		Handler: deployKeyReadFn,
	})
	Tool.RegisterWrite(server.ServerTool{
		Tool:    DeployKeyWriteTool,
		Handler: deployKeyWriteFn,
	})
}

func deployKeyReadFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	args := req.GetArguments()
	method, err := params.GetString(args, "method")
	if err != nil {
		return to.ErrorResult(err)
	}
	switch method {
	case "list":
		return listDeployKeysFn(ctx, req)
	case "get":
		return getDeployKeyFn(ctx, req)
	default:
		return to.ErrorResult(fmt.Errorf("unknown method: %s", method))
	}
}

func listDeployKeysFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	log.Debugf("Called listDeployKeysFn")
	args := req.GetArguments()
	owner, err := params.GetString(args, "owner")
	if err != nil {
		return to.ErrorResult(err)
	}
	repo, err := params.GetString(args, "repo")
	if err != nil {
		return to.ErrorResult(err)
	}
	page, pageSize := params.GetPagination(args, 30)
	client, err := gitea.ClientFromContext(ctx)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("get gitea client err: %v", err))
	}
	keys, _, err := client.ListDeployKeys(owner, repo, gitea_sdk.ListDeployKeysOptions{
		ListOptions: gitea_sdk.ListOptions{
			Page:     page,
			PageSize: pageSize,
		},
	})
	if err != nil {
		return to.ErrorResult(fmt.Errorf("list deploy keys %v/%v err: %v", owner, repo, err))
	}
	return to.TextResult(slimDeployKeys(keys))
}

func getDeployKeyFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	log.Debugf("Called getDeployKeyFn")
	args := req.GetArguments()
	owner, err := params.GetString(args, "owner")
	if err != nil {
		return to.ErrorResult(err)
	}
	repo, err := params.GetString(args, "repo")
	if err != nil {
		return to.ErrorResult(err)
	}
	id, err := params.GetIndex(args, "id")
	if err != nil {
		return to.ErrorResult(err)
	}
	client, err := gitea.ClientFromContext(ctx)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("get gitea client err: %v", err))
	}
	key, _, err := client.GetDeployKey(owner, repo, id)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("get deploy key %v in %v/%v err: %v", id, owner, repo, err))
	}
	return to.TextResult(slimDeployKey(key))
}

func deployKeyWriteFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	args := req.GetArguments()
	method, err := params.GetString(args, "method")
	if err != nil {
		return to.ErrorResult(err)
	}
	switch method {
	case "create":
		return createDeployKeyFn(ctx, req)
	case "delete":
		return deleteDeployKeyFn(ctx, req)
	default:
		return to.ErrorResult(fmt.Errorf("unknown method: %s", method))
	}
}

func createDeployKeyFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	log.Debugf("Called createDeployKeyFn")
	args := req.GetArguments()
	owner, err := params.GetString(args, "owner")
	if err != nil {
		return to.ErrorResult(err)
	}
	repo, err := params.GetString(args, "repo")
	if err != nil {
		return to.ErrorResult(err)
	}
	title, err := params.GetString(args, "title")
	if err != nil {
		return to.ErrorResult(err)
	}
	key, err := params.GetString(args, "key")
	if err != nil {
		return to.ErrorResult(err)
	}
	readOnly := true
	if v, ok := args["read_only"].(bool); ok {
		readOnly = v
	}
	client, err := gitea.ClientFromContext(ctx)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("get gitea client err: %v", err))
	}
	dk, _, err := client.CreateDeployKey(owner, repo, gitea_sdk.CreateKeyOption{
		Title:    title,
		Key:      key,
		ReadOnly: readOnly,
	})
	if err != nil {
		return to.ErrorResult(fmt.Errorf("create deploy key in %v/%v err: %v", owner, repo, err))
	}
	return to.TextResult(slimDeployKey(dk))
}

func deleteDeployKeyFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	log.Debugf("Called deleteDeployKeyFn")
	args := req.GetArguments()
	owner, err := params.GetString(args, "owner")
	if err != nil {
		return to.ErrorResult(err)
	}
	repo, err := params.GetString(args, "repo")
	if err != nil {
		return to.ErrorResult(err)
	}
	id, err := params.GetIndex(args, "id")
	if err != nil {
		return to.ErrorResult(err)
	}
	client, err := gitea.ClientFromContext(ctx)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("get gitea client err: %v", err))
	}
	_, err = client.DeleteDeployKey(owner, repo, id)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("delete deploy key %v from %v/%v err: %v", id, owner, repo, err))
	}
	return mcp.NewToolResultText(fmt.Sprintf("Deploy key %d deleted from %s/%s", id, owner, repo)), nil
}