Add webhook notifications and an MCP server

README.md

@@ -404,6 +404,7 @@ dde signal --input examples/signal-update.json
 dde backtest --input internal/backtest/testdata/scenario.json
 dde migrate
 dde serve
+dde mcp
 dde version
 ```
 
@@ -449,6 +450,47 @@ GET /v1/plans/{id}/versions
 
 ---
 
+## MCP server
+
+The engine is also an **MCP server**: every use-case above is exposed as a
+Model Context Protocol tool, so MCP-capable agents (Claude Code, Claude
+Desktop, custom runtimes) can create goals, generate ranked plans, submit
+signals, record outcomes and audit the version history with zero integration
+code.
+
+```bash
+dde mcp        # stdio — point Claude Code/Desktop at {"command": "dde", "args": ["mcp"]}
+dde serve      # also mounts streamable HTTP MCP at http://localhost:8080/mcp
+```
+
+Both transports share the engine's semantics with REST; the `/mcp` endpoint
+shares the live service with the API.
+
+> 📖 **Tool reference, transports and agent-loop guidance:** [`docs/mcp.md`](docs/mcp.md)
+
+---
+
+## Webhooks
+
+The engine can push domain events (`goal.created`, `plan.created`,
+`signal.received`, `replan.completed`, `outcome.recorded`,
+`goal.status_changed`) to an HTTP endpoint as they happen — Slack bridges,
+n8n/Zapier flows and agent triggers react to decisions without polling.
+Off by default; enable with:
+
+```bash
+DDE_WEBHOOK_URL=https://example.com/hook \
+DDE_WEBHOOK_SECRET=s3cret \
+dde serve
+```
+
+Deliveries are best-effort with retries and an HMAC-SHA256 signature header
+(`X-DDE-Signature`) for receiver-side verification.
+
+> 📖 **Event types, envelope format, signature verification and delivery semantics:** [`docs/api.md`](docs/api.md#webhooks)
+
+---
+
 ## Architecture
 
 ```text
@@ -493,6 +535,8 @@ The initial core is implemented and tested:
 * Postgres persistence with ordered migrations
 * Prometheus metrics + Grafana dashboards
 * a minimal Next.js admin UI
+* webhook event notifications
+* an MCP server (stdio + streamable HTTP)
 
 More LLM providers, local / self-hosted inference, authentication, richer scoring, OpenTelemetry, and deeper admin workflows are planned next.
 
@@ -524,6 +568,8 @@ More LLM providers, local / self-hosted inference, authentication, richer scorin
 * [x] Async, FIFO-per-plan replanning + durable signal status + plan cache (TTL for finance)
 * [x] Per-domain metrics
 * [x] Pluggable external data sources (HTTP/MCP/AI-agent/read-model; deterministic pre-fetch, provenance-tracked)
+* [x] Webhook event notifications (best-effort, HMAC-signed, retried)
+* [x] MCP server (stdio via `dde mcp` + streamable HTTP at `/mcp`)
 * [ ] Local / self-hosted LLM inference (OpenAI-compatible endpoint, provider TBD)
 * [ ] Real market-data vendor (HTTP provider stub in place)
 * [ ] OpenTelemetry tracing

cmd/dde/commands.go

@@ -4,11 +4,14 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"log/slog"
+	"net/http"
 	"os"
 	"os/signal"
 	"syscall"
 	"time"
 
+	"github.com/modelcontextprotocol/go-sdk/mcp"
 	"github.com/spf13/cobra"
 
 	"github.com/vingrad/dynamic-decision-engine/internal/api"
@@ -20,9 +23,11 @@ import (
 	"github.com/vingrad/dynamic-decision-engine/internal/llm"
 	"github.com/vingrad/dynamic-decision-engine/internal/logging"
 	"github.com/vingrad/dynamic-decision-engine/internal/marketdata"
+	mcpserver "github.com/vingrad/dynamic-decision-engine/internal/mcp"
 	"github.com/vingrad/dynamic-decision-engine/internal/pack"
 	"github.com/vingrad/dynamic-decision-engine/internal/policy"
 	"github.com/vingrad/dynamic-decision-engine/internal/storage"
+	"github.com/vingrad/dynamic-decision-engine/internal/webhook"
 	"github.com/vingrad/dynamic-decision-engine/internal/wire"
 )
 
@@ -204,6 +209,87 @@ func newEngine(cfg config.Config, reg *pack.Registry, pol policy.Policy, cacheOb
 	return engine.New(router, opts...), nil
 }
 
+// buildService opens storage and assembles the full production service: domain
+// registry, policy, engine, metrics, optional webhook notifier and optional
+// async replan queue (including crash recovery). It is shared by the long-running
+// transports (serve, mcp). The returned cleanup drains the service (in-flight
+// replans first, then queued webhook deliveries — replans emit events) and
+// closes the repository; call it on shutdown.
+func buildService(ctx context.Context, cfg config.Config, log *slog.Logger, metrics *api.Metrics) (*app.Service, func(), error) {
+	repo, err := storage.Open(ctx, storage.Options{
+		DatabaseURL: cfg.DatabaseURL,
+		MaxConns:    cfg.DBMaxConns,
+	}, log)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	reg, err := buildRegistry(cfg)
+	if err != nil {
+		repo.Close()
+		return nil, nil, err
+	}
+	pol, err := policy.Load(cfg.PolicyFile)
+	if err != nil {
+		repo.Close()
+		return nil, nil, err
+	}
+	eng, err := newEngine(cfg, reg, pol, metrics)
+	if err != nil {
+		repo.Close()
+		return nil, nil, err
+	}
+
+	opts := []app.Option{app.WithMetrics(metrics), app.WithLogger(log), app.WithRegistry(reg)}
+	var hooks *webhook.Dispatcher
+	if cfg.WebhookURL != "" {
+		hooks = webhook.New(webhook.Config{
+			URL:     cfg.WebhookURL,
+			Secret:  cfg.WebhookSecret,
+			Timeout: cfg.WebhookTimeout,
+			Retries: cfg.WebhookRetries,
+			Events:  cfg.WebhookEvents,
+		}, log, metrics)
+		opts = append(opts, app.WithNotifier(hooks))
+	}
+	if cfg.ReplanAsync {
+		opts = append(opts,
+			app.WithReplanQueue(app.NewMemoryQueue(cfg.ReplanWorkers, 1024, log, app.WithQueueTimeout(cfg.ReplanTimeout))),
+			app.WithReplanRetries(cfg.ReplanMaxRetries),
+		)
+	}
+	svc := app.New(repo, eng, opts...)
+	// Re-enqueue replans left pending by a previous crash (async only — the
+	// in-memory queue loses scheduled work on restart; inline never does).
+	if cfg.ReplanAsync {
+		go func() {
+			n, err := svc.RecoverPending(ctx)
+			if err != nil {
+				log.Error("replan recovery failed", "err", err)
+				return
+			}
+			if n > 0 {
+				log.Info("recovered pending replans", "count", n)
+			}
+		}()
+	}
+
+	cleanup := func() {
+		// Drain in-flight async replanning first: completing replans emit events,
+		// which the webhook drain below then delivers.
+		sctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		_ = svc.Shutdown(sctx)
+		cancel()
+		if hooks != nil {
+			wctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+			_ = hooks.Shutdown(wctx)
+			cancel()
+		}
+		repo.Close()
+	}
+	return svc, cleanup, nil
+}
+
 // newServeCommand runs the REST API server.
 func newServeCommand() *cobra.Command {
 	return &cobra.Command{
@@ -220,59 +306,18 @@ func newServeCommand() *cobra.Command {
 			ctx, stop := signal.NotifyContext(cmd.Context(), syscall.SIGINT, syscall.SIGTERM)
 			defer stop()
 
-			repo, err := storage.Open(ctx, storage.Options{
-				DatabaseURL: cfg.DatabaseURL,
-				MaxConns:    cfg.DBMaxConns,
-			}, log)
-			if err != nil {
-				return err
-			}
-			defer repo.Close()
-
-			reg, err := buildRegistry(cfg)
-			if err != nil {
-				return err
-			}
-			pol, err := policy.Load(cfg.PolicyFile)
-			if err != nil {
-				return err
-			}
 			metrics := api.NewMetrics()
-			eng, err := newEngine(cfg, reg, pol, metrics)
+			svc, cleanup, err := buildService(ctx, cfg, log, metrics)
 			if err != nil {
 				return err
 			}
+			defer cleanup()
 
-			opts := []app.Option{app.WithMetrics(metrics), app.WithLogger(log), app.WithRegistry(reg)}
-			if cfg.ReplanAsync {
-				opts = append(opts,
-					app.WithReplanQueue(app.NewMemoryQueue(cfg.ReplanWorkers, 1024, log, app.WithQueueTimeout(cfg.ReplanTimeout))),
-					app.WithReplanRetries(cfg.ReplanMaxRetries),
-				)
-			}
-			svc := app.New(repo, eng, opts...)
-			// Re-enqueue replans left pending by a previous crash (async only — the
-			// in-memory queue loses scheduled work on restart; inline never does).
-			if cfg.ReplanAsync {
-				go func() {
-					n, err := svc.RecoverPending(ctx)
-					if err != nil {
-						log.Error("replan recovery failed", "err", err)
-						return
-					}
-					if n > 0 {
-						log.Info("recovered pending replans", "count", n)
-					}
-				}()
-			}
-			// Drain in-flight async replanning on shutdown.
-			defer func() {
-				sctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-				defer cancel()
-				_ = svc.Shutdown(sctx)
-			}()
-
-			srv := api.New(cfg, log, svc, metrics)
+			// The MCP endpoint shares the live service: REST and MCP clients see
+			// the same goals, plans and versions.
+			mcpSrv := mcpserver.New(svc, version)
+			srv := api.New(cfg, log, svc, metrics,
+				api.WithMCP(mcp.NewStreamableHTTPHandler(func(*http.Request) *mcp.Server { return mcpSrv }, nil)))
 			return srv.Run(ctx)
 		},
 	}

cmd/dde/main.go

@@ -26,6 +26,7 @@ func main() {
 
 	root.AddCommand(
 		newServeCommand(),
+		newMCPCommand(),
 		newMigrateCommand(),
 		newEvaluateCommand(),
 		newSignalCommand(),

cmd/dde/mcp.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	"os/signal"
+	"syscall"
+
+	"github.com/modelcontextprotocol/go-sdk/mcp"
+	"github.com/spf13/cobra"
+
+	"github.com/vingrad/dynamic-decision-engine/internal/api"
+	"github.com/vingrad/dynamic-decision-engine/internal/config"
+	"github.com/vingrad/dynamic-decision-engine/internal/logging"
+	mcpserver "github.com/vingrad/dynamic-decision-engine/internal/mcp"
+)
+
+// newMCPCommand serves the engine as an MCP server over stdio, so MCP clients
+// (Claude Code/Desktop, agent runtimes) can drive the decision loop as tools.
+// It assembles the same production service as serve — same DDE_* config, store
+// (in-memory by default, Postgres via DATABASE_URL), planner and webhooks.
+//
+// Stdout carries JSON-RPC only: logging.New writes to stderr, and nothing in
+// this path may print to stdout.
+func newMCPCommand() *cobra.Command {
+	return &cobra.Command{
+		Use:   "mcp",
+		Short: "Serve the engine as an MCP server over stdio",
+		Long: "mcp exposes the engine's use-cases (evaluate, create_goal, generate_plan, " +
+			"submit_signal, record_outcome, ...) as Model Context Protocol tools over stdio. " +
+			"Configuration matches serve: DDE_* environment variables select the store, " +
+			"planner and webhooks.",
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			cfg, err := config.Load()
+			if err != nil {
+				return err
+			}
+			log := logging.New(cfg.LogLevel, cfg.LogFormat)
+
+			ctx, stop := signal.NotifyContext(cmd.Context(), syscall.SIGINT, syscall.SIGTERM)
+			defer stop()
+
+			// Metrics are recorded but not exported in stdio mode (no HTTP listener).
+			metrics := api.NewMetrics()
+			svc, cleanup, err := buildService(ctx, cfg, log, metrics)
+			if err != nil {
+				return err
+			}
+			defer cleanup()
+
+			srv := mcpserver.New(svc, version)
+			return srv.Run(ctx, &mcp.StdioTransport{})
+		},
+	}
+}