Tyk AI Studio Custom Endpoint Plugins

/plugins/{slug}/{path...}

```mermaid
sequenceDiagram
    participant Client as HTTP Client
    participant Gateway as Edge Gateway (Gin Router)
    participant Plugin as Plugin Process

    Client->>Gateway: POST /plugins/my-mcp/mcp
    Note over Gateway: 1. Match /plugins/*path catch-all route<br/>2. Parse slug + relative path<br/>3. Look up EndpointRoute<br/>4. Validate auth & fetch App<br/>5. Build EndpointRequest
    Gateway->>Plugin: gRPC HandleEndpointRequest()
    Note over Plugin: CustomEndpointHandler<br/>Full control over response<br/>Access to App metadata
    Plugin-->>Gateway: EndpointResponse
    Gateway-->>Client: HTTP Response

type CustomEndpointHandler interface {
    Plugin
    GetEndpointRegistrations() ([]*pb.EndpointRegistration, error)
    HandleEndpointRequest(ctx Context, req *pb.EndpointRequest) (*pb.EndpointResponse, error)
    HandleEndpointRequestStream(ctx Context, req *pb.EndpointRequest, stream grpc.ServerStreamingServer[pb.EndpointResponseChunk]) error
}

func (p *MyPlugin) GetEndpointRegistrations() ([]*pb.EndpointRegistration, error) {
    return []*pb.EndpointRegistration{
        {
            Path:           "/*",                            // Catch-all wildcard
            Methods:        []string{"GET", "POST", "DELETE"},
            Description:    "MCP Streamable HTTP endpoint",
            RequireAuth:    true,                            // Gateway enforces auth
            StreamResponse: true,                            // Use streaming RPC
        },
    }, nil
}

func (p *MyPlugin) HandleEndpointRequest(ctx plugin_sdk.Context, req *pb.EndpointRequest) (*pb.EndpointResponse, error) {
    // Route based on path segments
    segments := req.PathSegments

    switch {
    case len(segments) == 0:
        return p.handleRoot(req)
    case segments[0] == ".well-known" && len(segments) >= 2:
        return p.handleWellKnown(segments[1], req)
    case segments[0] == "users" && len(segments) >= 2:
        return p.handleUser(segments[1], req)
    default:
        return &pb.EndpointResponse{
            StatusCode: 404,
            Headers:    map[string]string{"Content-Type": "application/json"},
            Body:       []byte(`{"error": "not found"}`),
        }, nil
    }
}

func (p *MyPlugin) HandleEndpointRequestStream(
    ctx plugin_sdk.Context,
    req *pb.EndpointRequest,
    stream grpc.ServerStreamingServer[pb.EndpointResponseChunk],
) error {
    // Send headers first
    stream.Send(&pb.EndpointResponseChunk{
        Type:       pb.EndpointResponseChunk_HEADERS,
        StatusCode: 200,
        Headers: map[string]string{
            "Content-Type":  "text/event-stream",
            "Cache-Control": "no-cache",
            "Connection":    "keep-alive",
        },
    })

    // Send SSE data chunks — each is flushed immediately to the HTTP client
    for i := 0; i < 5; i++ {
        data := fmt.Sprintf("data: {\"count\": %d}\n\n", i)
        stream.Send(&pb.EndpointResponseChunk{
            Type: pb.EndpointResponseChunk_BODY,
            Data: []byte(data),
        })
        time.Sleep(1 * time.Second)
    }

    // Signal stream completion
    stream.Send(&pb.EndpointResponseChunk{
        Type: pb.EndpointResponseChunk_DONE,
    })

    return nil
}

{
  "id": "com.example.my-plugin",
  "name": "My Custom Endpoint",
  "version": "1.0.0",
  "capabilities": {
    "hooks": ["custom_endpoint"],
    "primary_hook": "custom_endpoint"
  }
}

"hooks": ["custom_endpoint", "studio_ui"]

Plugin has endpoint registrations but no 'slug' in config

# Build the plugin
cd examples/plugins/gateway/custom-echo-endpoint
go build -o custom-echo-endpoint

# Register with file:// for local development
# Command: file:///path/to/custom-echo-endpoint
# Config: {"slug": "custom-echo-endpoint", "custom_content": "Hello World"}

func (p *MyPlugin) HandleEndpointRequest(ctx plugin_sdk.Context, req *pb.EndpointRequest) (*pb.EndpointResponse, error) {
    if !req.Authenticated || req.App == nil {
        return &pb.EndpointResponse{StatusCode: 401, Body: []byte("Unauthorized")}, nil
    }

    // Check custom ACL in App metadata
    if req.App.Metadata["mcp_access"] != "allowed" {
        return &pb.EndpointResponse{StatusCode: 403, Body: []byte("Forbidden")}, nil
    }

    // Check allowed operations
    allowedOps := req.App.Metadata["allowed_operations"]
    if allowedOps != "" && !strings.Contains(allowedOps, req.Method) {
        return &pb.EndpointResponse{StatusCode: 405, Body: []byte("Method not allowed for this app")}, nil
    }

    // Access granted — handle request
    return p.processRequest(req)
}

func (p *MyPlugin) GetEndpointRegistrations() ([]*pb.EndpointRegistration, error) {
    return []*pb.EndpointRegistration{
        {
            Path:        "/.well-known/openid-configuration",
            Methods:     []string{"GET"},
            Description: "OpenID Connect discovery",
        },
        {
            Path:        "/token",
            Methods:     []string{"POST"},
            Description: "Token endpoint",
            RequireAuth: false,
        },
        {
            Path:        "/userinfo",
            Methods:     []string{"GET"},
            Description: "UserInfo endpoint",
            RequireAuth: true,
        },
    }, nil
}

type MCPProxyPlugin struct {
    plugin_sdk.BasePlugin
    sessions sync.Map // sessionID → session state
}

func (p *MCPProxyPlugin) GetEndpointRegistrations() ([]*pb.EndpointRegistration, error) {
    return []*pb.EndpointRegistration{
        {
            Path:           "/*",
            Methods:        []string{"POST", "GET", "DELETE"},
            Description:    "MCP Streamable HTTP proxy",
            RequireAuth:    true,
            StreamResponse: true, // Use streaming for all requests
        },
    }, nil
}

func (p *MCPProxyPlugin) HandleEndpointRequestStream(
    ctx plugin_sdk.Context,
    req *pb.EndpointRequest,
    stream grpc.ServerStreamingServer[pb.EndpointResponseChunk],
) error {
    switch req.Method {
    case "POST":
        return p.handleMCPPost(ctx, req, stream)
    case "GET":
        return p.handleMCPGet(ctx, req, stream)
    case "DELETE":
        return p.handleMCPDelete(ctx, req, stream)
    default:
        stream.Send(&pb.EndpointResponseChunk{
            Type:       pb.EndpointResponseChunk_HEADERS,
            StatusCode: 405,
        })
        stream.Send(&pb.EndpointResponseChunk{Type: pb.EndpointResponseChunk_DONE})
        return nil
    }
}

func (p *MCPProxyPlugin) handleMCPPost(ctx plugin_sdk.Context, req *pb.EndpointRequest, stream grpc.ServerStreamingServer[pb.EndpointResponseChunk]) error {
    // Parse JSON-RPC request
    var rpcReq map[string]interface{}
    json.Unmarshal(req.Body, &rpcReq)

    // Check if client accepts SSE
    acceptsSSE := strings.Contains(req.Headers["Accept"], "text/event-stream")

    if acceptsSSE {
        // Stream response as SSE
        stream.Send(&pb.EndpointResponseChunk{
            Type:       pb.EndpointResponseChunk_HEADERS,
            StatusCode: 200,
            Headers: map[string]string{
                "Content-Type":  "text/event-stream",
                "Cache-Control": "no-cache",
            },
        })

        // Send SSE events...
        for _, event := range p.processRPCRequest(rpcReq) {
            data := fmt.Sprintf("data: %s\n\n", event)
            stream.Send(&pb.EndpointResponseChunk{
                Type: pb.EndpointResponseChunk_BODY,
                Data: []byte(data),
            })
        }
    } else {
        // Single JSON response
        result := p.processRPCRequestSync(rpcReq)
        body, _ := json.Marshal(result)
        stream.Send(&pb.EndpointResponseChunk{
            Type:       pb.EndpointResponseChunk_HEADERS,
            StatusCode: 200,
            Headers:    map[string]string{"Content-Type": "application/json"},
        })
        stream.Send(&pb.EndpointResponseChunk{
            Type: pb.EndpointResponseChunk_BODY,
            Data: body,
        })
    }

    stream.Send(&pb.EndpointResponseChunk{Type: pb.EndpointResponseChunk_DONE})
    return nil
}

package main

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "encoding/json"
    "fmt"

    "github.com/TykTechnologies/midsommar/v2/pkg/plugin_sdk"
    pb "github.com/TykTechnologies/midsommar/v2/proto"
)

type WebhookPlugin struct {
    plugin_sdk.BasePlugin
    secret string
}

func NewWebhookPlugin() *WebhookPlugin {
    return &WebhookPlugin{
        BasePlugin: plugin_sdk.NewBasePlugin("webhook-receiver", "1.0.0", "Receives and validates webhooks"),
    }
}

func (p *WebhookPlugin) Initialize(ctx plugin_sdk.Context, config map[string]string) error {
    p.secret = config["webhook_secret"]
    if p.secret == "" {
        return fmt.Errorf("webhook_secret is required")
    }
    ctx.Services.Logger().Info("Webhook receiver initialized")
    return nil
}

func (p *WebhookPlugin) GetEndpointRegistrations() ([]*pb.EndpointRegistration, error) {
    return []*pb.EndpointRegistration{
        {
            Path:        "/*",
            Methods:     []string{"POST"},
            Description: "Webhook receiver endpoint",
            RequireAuth: false, // Webhooks use their own auth (HMAC signature)
        },
    }, nil
}

func (p *WebhookPlugin) HandleEndpointRequest(ctx plugin_sdk.Context, req *pb.EndpointRequest) (*pb.EndpointResponse, error) {
    // Validate HMAC signature
    signature := req.Headers["X-Webhook-Signature"]
    if !p.validateSignature(req.Body, signature) {
        return &pb.EndpointResponse{
            StatusCode: 401,
            Headers:    map[string]string{"Content-Type": "application/json"},
            Body:       []byte(`{"error": "invalid signature"}`),
        }, nil
    }

    // Route by path segments
    segments := req.PathSegments
    if len(segments) == 0 {
        return &pb.EndpointResponse{StatusCode: 400, Body: []byte(`{"error": "missing event type"}`)}, nil
    }

    eventType := segments[0]
    ctx.Services.Logger().Info("Webhook received", "event", eventType, "body_size", len(req.Body))

    // Process webhook
    switch eventType {
    case "payment":
        return p.handlePaymentWebhook(ctx, req)
    case "user":
        return p.handleUserWebhook(ctx, req)
    default:
        return &pb.EndpointResponse{
            StatusCode: 200,
            Headers:    map[string]string{"Content-Type": "application/json"},
            Body:       []byte(`{"status": "acknowledged"}`),
        }, nil
    }
}

func (p *WebhookPlugin) HandleEndpointRequestStream(ctx plugin_sdk.Context, req *pb.EndpointRequest, stream pb.PluginService_HandleEndpointRequestStreamServer) error {
    return fmt.Errorf("streaming not supported for webhooks")
}

func (p *WebhookPlugin) validateSignature(body []byte, signature string) bool {
    mac := hmac.New(sha256.New, []byte(p.secret))
    mac.Write(body)
    expected := hex.EncodeToString(mac.Sum(nil))
    return hmac.Equal([]byte(expected), []byte(signature))
}

func main() {
    plugin_sdk.Serve(NewWebhookPlugin())
}

PLUGIN_ENDPOINT_STREAM_TIMEOUT=30m