gitea-mcp/operation/repo/webhook.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 (
	WebhookReadToolName  = "webhook_read"
	WebhookWriteToolName = "webhook_write"
)

var (
	WebhookReadTool = mcp.NewTool(
		WebhookReadToolName,
		mcp.WithDescription("Read repository webhooks. Use method 'list' to list all, 'get' to get a specific webhook."),
		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("webhook 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)),
	)

	WebhookWriteTool = mcp.NewTool(
		WebhookWriteToolName,
		mcp.WithDescription("Create, edit, delete, or test repository webhooks."),
		mcp.WithString("method", mcp.Required(), mcp.Description("operation to perform"), mcp.Enum("create", "edit", "delete", "test")),
		mcp.WithString("owner", mcp.Required(), mcp.Description("repository owner")),
		mcp.WithString("repo", mcp.Required(), mcp.Description("repository name")),
		mcp.WithNumber("id", mcp.Description("webhook ID (required for 'edit', 'delete', 'test')")),
		mcp.WithString("type", mcp.Description("webhook type (for 'create')"), mcp.Enum("gitea", "slack", "discord", "dingtalk", "telegram", "msteams", "feishu", "wechatwork", "packagist")),
		mcp.WithString("url", mcp.Description("target URL (required for 'create', optional for 'edit')")),
		mcp.WithString("content_type", mcp.Description("content type"), mcp.Enum("json", "form")),
		mcp.WithString("secret", mcp.Description("webhook secret")),
		mcp.WithArray("events", mcp.Description("event types to trigger on (e.g. push, pull_request, issues)"), mcp.Items(map[string]any{"type": "string"})),
		mcp.WithBoolean("active", mcp.Description("whether the webhook is active")),
		mcp.WithString("branch_filter", mcp.Description("branch filter pattern")),
	)
)

func init() {
	Tool.RegisterRead(server.ServerTool{
		Tool:    WebhookReadTool,
		Handler: webhookReadFn,
	})
	Tool.RegisterWrite(server.ServerTool{
		Tool:    WebhookWriteTool,
		Handler: webhookWriteFn,
	})
}

func webhookReadFn(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 listWebhooksFn(ctx, req)
	case "get":
		return getWebhookFn(ctx, req)
	default:
		return to.ErrorResult(fmt.Errorf("unknown method: %s", method))
	}
}

func listWebhooksFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	log.Debugf("Called listWebhooksFn")
	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))
	}
	hooks, _, err := client.ListRepoHooks(owner, repo, gitea_sdk.ListHooksOptions{
		ListOptions: gitea_sdk.ListOptions{
			Page:     page,
			PageSize: pageSize,
		},
	})
	if err != nil {
		return to.ErrorResult(fmt.Errorf("list webhooks %v/%v err: %v", owner, repo, err))
	}
	return to.TextResult(slimHooks(hooks))
}

func getWebhookFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	log.Debugf("Called getWebhookFn")
	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))
	}
	hook, _, err := client.GetRepoHook(owner, repo, id)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("get webhook %v in %v/%v err: %v", id, owner, repo, err))
	}
	return to.TextResult(slimHook(hook))
}

func webhookWriteFn(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 createWebhookFn(ctx, req)
	case "edit":
		return editWebhookFn(ctx, req)
	case "delete":
		return deleteWebhookFn(ctx, req)
	case "test":
		return testWebhookFn(ctx, req)
	default:
		return to.ErrorResult(fmt.Errorf("unknown method: %s", method))
	}
}

func createWebhookFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	log.Debugf("Called createWebhookFn")
	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)
	}
	hookType := params.GetOptionalString(args, "type", "gitea")
	url, err := params.GetString(args, "url")
	if err != nil {
		return to.ErrorResult(err)
	}
	contentType := params.GetOptionalString(args, "content_type", "json")
	secret, _ := args["secret"].(string)
	events := params.GetStringSlice(args, "events")
	active := true
	if v, ok := args["active"].(bool); ok {
		active = v
	}
	branchFilter, _ := args["branch_filter"].(string)

	client, err := gitea.ClientFromContext(ctx)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("get gitea client err: %v", err))
	}

	opt := gitea_sdk.CreateHookOption{
		Type: gitea_sdk.HookType(hookType),
		Config: map[string]string{
			"url":          url,
			"content_type": contentType,
			"secret":       secret,
		},
		Events:       events,
		Active:       active,
		BranchFilter: branchFilter,
	}

	hook, _, err := client.CreateRepoHook(owner, repo, opt)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("create webhook in %v/%v err: %v", owner, repo, err))
	}
	return to.TextResult(slimHook(hook))
}

func editWebhookFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	log.Debugf("Called editWebhookFn")
	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))
	}

	opt := gitea_sdk.EditHookOption{}
	config := map[string]string{}
	if v, ok := args["url"].(string); ok {
		config["url"] = v
	}
	if v, ok := args["content_type"].(string); ok {
		config["content_type"] = v
	}
	if v, ok := args["secret"].(string); ok {
		config["secret"] = v
	}
	if len(config) > 0 {
		opt.Config = config
	}
	if events := params.GetStringSlice(args, "events"); events != nil {
		opt.Events = events
	}
	if v, ok := args["active"].(bool); ok {
		opt.Active = &v
	}
	if v, ok := args["branch_filter"].(string); ok {
		opt.BranchFilter = v
	}

	_, err = client.EditRepoHook(owner, repo, id, opt)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("edit webhook %v in %v/%v err: %v", id, owner, repo, err))
	}
	// EditRepoHook doesn't return the hook, fetch it
	hook, _, err := client.GetRepoHook(owner, repo, id)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("get webhook %v in %v/%v after edit err: %v", id, owner, repo, err))
	}
	return to.TextResult(slimHook(hook))
}

func deleteWebhookFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	log.Debugf("Called deleteWebhookFn")
	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.DeleteRepoHook(owner, repo, id)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("delete webhook %v from %v/%v err: %v", id, owner, repo, err))
	}
	return mcp.NewToolResultText(fmt.Sprintf("Webhook %d deleted from %s/%s", id, owner, repo)), nil
}

func testWebhookFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	log.Debugf("Called testWebhookFn")
	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)
	}
	// TestRepoHook is not in the SDK - use raw REST
	_, err = gitea.DoJSON(ctx, "POST", fmt.Sprintf("repos/%s/%s/hooks/%d/tests", owner, repo, id), nil, nil, nil)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("test webhook %v in %v/%v err: %v", id, owner, repo, err))
	}
	return mcp.NewToolResultText(fmt.Sprintf("Webhook %d test triggered for %s/%s", id, owner, repo)), nil
}