feat: sqlite persistence with goose migrations, flashcard schema, zap logger
- Add SQLite as authoritative store alongside Qdrant (best-effort indexing) - Use goose for embedded SQL migrations - Schema matches flashcard format: id(INTEGER PK), note, repos, host, tags, user, context, created - Replace std log with zap structured logger - Split registerTools into registerStoreKnowledge + registerSearchKnowledge - Qdrant index uses SQLite rowid directly as point ID (no SHA256 hash) - Search enriches Qdrant results from SQLite by ID - Lowercase auth header before prefix check - Always handle json.MarshalIndent errors - NixOS: StateDirectory, sqlitePath to /var/lib/st-pierre/st-pierre.sqlite
This commit is contained in:
1 parent
5ac11631d4
commit
39c6185b70
8 files changed
+366
-94
No files matched your search
@@ -4,7 +4,6 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
@@ -16,14 +15,21 @@ import (
|
||||
"github.com/mark3labs/mcp-go/mcp"
|
||||
"github.com/mark3labs/mcp-go/server"
|
||||
"github.com/qdrant/go-client/qdrant"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
var version = "0.1.0"
|
||||
|
||||
func main() {
|
||||
qdrantURL := requireEnv("QDRANT_URL")
|
||||
openAIKey := requireEnv("OPENAI_API_KEY")
|
||||
token := requireEnv("MCP_TOKEN")
|
||||
logger, err := zap.NewProduction()
|
||||
if err != nil {
|
||||
panic(fmt.Sprintf("failed to init logger: %v", err))
|
||||
}
|
||||
defer logger.Sync()
|
||||
|
||||
qdrantURL := requireEnv(logger, "QDRANT_URL")
|
||||
openAIKey := requireEnv(logger, "OPENAI_API_KEY")
|
||||
token := requireEnv(logger, "MCP_TOKEN")
|
||||
embeddingModel := os.Getenv("EMBEDDING_MODEL")
|
||||
embeddingBaseURL := os.Getenv("EMBEDDING_BASE_URL")
|
||||
port := envOrDefault("PORT", "8080")
|
||||
@@ -42,15 +48,22 @@ func main() {
|
||||
UseTLS: qdrantTLS,
|
||||
})
|
||||
if err != nil {
|
||||
log.Fatalf("failed to create qdrant client: %v", err)
|
||||
logger.Fatal("failed to create qdrant client", zap.Error(err))
|
||||
}
|
||||
defer qdrantClient.Close()
|
||||
|
||||
emb := newEmbedder(openAIKey, embeddingModel, embeddingBaseURL)
|
||||
store := newQdrantStore(qdrantClient, emb)
|
||||
qdrantStore := newQdrantStore(qdrantClient, emb)
|
||||
|
||||
if err := store.ensureCollection(context.Background()); err != nil {
|
||||
log.Fatalf("failed to ensure collection: %v", err)
|
||||
sqlitePath := envOrDefault("SQLITE_PATH", "knowledge.db")
|
||||
sqliteStore, err := newSqliteStore(sqlitePath)
|
||||
if err != nil {
|
||||
logger.Fatal("failed to init sqlite", zap.Error(err))
|
||||
}
|
||||
defer sqliteStore.close()
|
||||
|
||||
if err := qdrantStore.ensureCollection(context.Background()); err != nil {
|
||||
logger.Fatal("failed to ensure collection", zap.Error(err))
|
||||
}
|
||||
|
||||
mcpServer := server.NewMCPServer(
|
||||
@@ -60,7 +73,8 @@ func main() {
|
||||
server.WithLogging(),
|
||||
)
|
||||
|
||||
registerTools(mcpServer, store)
|
||||
registerStoreKnowledge(mcpServer, sqliteStore, qdrantStore, logger)
|
||||
registerSearchKnowledge(mcpServer, sqliteStore, qdrantStore, logger)
|
||||
|
||||
streamableServer := server.NewStreamableHTTPServer(mcpServer)
|
||||
|
||||
@@ -71,42 +85,57 @@ func main() {
|
||||
Handler: corsMiddleware(authHandler),
|
||||
}
|
||||
|
||||
log.Printf("MCP Knowledge Server starting on http://0.0.0.0:%s", port)
|
||||
logger.Info("starting", zap.String("port", port))
|
||||
|
||||
go func() {
|
||||
sigChan := make(chan os.Signal, 1)
|
||||
signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
|
||||
sig := <-sigChan
|
||||
log.Printf("received signal %v, shutting down...", sig)
|
||||
logger.Info("shutting down", zap.String("signal", sig.String()))
|
||||
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
if err := streamableServer.Shutdown(shutdownCtx); err != nil {
|
||||
log.Printf("shutdown error: %v", err)
|
||||
logger.Error("shutdown error", zap.Error(err))
|
||||
}
|
||||
qdrantClient.Close()
|
||||
sqliteStore.close()
|
||||
os.Exit(0)
|
||||
}()
|
||||
|
||||
if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
||||
log.Fatalf("server stopped: %v", err)
|
||||
logger.Fatal("server stopped", zap.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
func registerTools(mcpServer *server.MCPServer, store *qdrantStore) {
|
||||
func registerStoreKnowledge(mcpServer *server.MCPServer, sqlite *sqliteStore, qdrant *qdrantStore, logger *zap.Logger) {
|
||||
mcpServer.AddTool(
|
||||
mcp.NewTool("store_knowledge",
|
||||
mcp.WithDescription("Store a piece of knowledge for later retrieval. Use this to persist useful information discovered during a session."),
|
||||
mcp.WithString("content",
|
||||
mcp.WithString("note",
|
||||
mcp.Required(),
|
||||
mcp.Description("The text content to store"),
|
||||
mcp.Description("The markdown body — keep it concise, one paragraph usually"),
|
||||
),
|
||||
mcp.WithObject("metadata",
|
||||
mcp.Description("Optional metadata as key-value pairs (e.g. {\"source\": \"docs\", \"category\": \"golang\"})"),
|
||||
mcp.WithArray("repos",
|
||||
mcp.Description("Optional list of repository names"),
|
||||
),
|
||||
mcp.WithString("host",
|
||||
mcp.Description("Optional hostname or null if universal"),
|
||||
),
|
||||
mcp.WithArray("tags",
|
||||
mcp.Description("Optional list of tags"),
|
||||
),
|
||||
mcp.WithString("user",
|
||||
mcp.Description("Optional user identifier"),
|
||||
),
|
||||
mcp.WithString("context",
|
||||
mcp.Description("Optional context — what triggered this"),
|
||||
),
|
||||
),
|
||||
handleStoreKnowledge(store),
|
||||
handleStoreKnowledge(sqlite, qdrant, logger),
|
||||
)
|
||||
}
|
||||
|
||||
func registerSearchKnowledge(mcpServer *server.MCPServer, sqlite *sqliteStore, qdrant *qdrantStore, logger *zap.Logger) {
|
||||
mcpServer.AddTool(
|
||||
mcp.NewTool("search_knowledge",
|
||||
mcp.WithDescription("Search previously stored knowledge using semantic search. Returns the most relevant knowledge entries for the given query."),
|
||||
@@ -118,42 +147,52 @@ func registerTools(mcpServer *server.MCPServer, store *qdrantStore) {
|
||||
mcp.Description("Maximum number of results to return (default: 5)"),
|
||||
),
|
||||
),
|
||||
handleSearchKnowledge(store),
|
||||
handleSearchKnowledge(sqlite, qdrant),
|
||||
)
|
||||
}
|
||||
|
||||
func handleStoreKnowledge(store *qdrantStore) server.ToolHandlerFunc {
|
||||
func handleStoreKnowledge(sqlite *sqliteStore, qdrant *qdrantStore, logger *zap.Logger) server.ToolHandlerFunc {
|
||||
return func(ctx context.Context, request mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
||||
args, ok := request.Params.Arguments.(map[string]any)
|
||||
if !ok {
|
||||
return mcp.NewToolResultError("invalid arguments"), nil
|
||||
}
|
||||
|
||||
content, ok := args["content"].(string)
|
||||
if !ok || content == "" {
|
||||
return mcp.NewToolResultError("content is required and must be a non-empty string"), nil
|
||||
note, ok := args["note"].(string)
|
||||
if !ok || note == "" {
|
||||
return mcp.NewToolResultError("note is required and must be a non-empty string"), nil
|
||||
}
|
||||
|
||||
var metadata map[string]any
|
||||
if rawMeta, ok := args["metadata"]; ok && rawMeta != nil {
|
||||
metaMap, ok := rawMeta.(map[string]any)
|
||||
if !ok {
|
||||
return mcp.NewToolResultError("metadata must be a key-value object"), nil
|
||||
}
|
||||
metadata = metaMap
|
||||
}
|
||||
repos := parseStringSlice(args["repos"])
|
||||
host := parseOptionalString(args["host"])
|
||||
tags := parseStringSlice(args["tags"])
|
||||
user := parseOptionalString(args["user"])
|
||||
context_ := parseOptionalString(args["context"])
|
||||
|
||||
result, err := store.store(ctx, content, metadata)
|
||||
id, err := sqlite.save(ctx, note, host, user, context_, repos, tags)
|
||||
if err != nil {
|
||||
return mcp.NewToolResultError(fmt.Sprintf("failed to store knowledge: %v", err)), nil
|
||||
return mcp.NewToolResultError(fmt.Sprintf("failed to persist: %v", err)), nil
|
||||
}
|
||||
|
||||
if err := qdrant.index(ctx, id, note); err != nil {
|
||||
logger.Warn("qdrant index failed (entry in sqlite)", zap.Int64("id", id), zap.Error(err))
|
||||
}
|
||||
|
||||
entry, err := sqlite.getByID(ctx, id)
|
||||
if err != nil {
|
||||
return mcp.NewToolResultError(fmt.Sprintf("failed to retrieve: %v", err)), nil
|
||||
}
|
||||
|
||||
output, err := json.MarshalIndent(entry, "", " ")
|
||||
if err != nil {
|
||||
return mcp.NewToolResultError(fmt.Sprintf("failed to marshal result: %v", err)), nil
|
||||
}
|
||||
|
||||
output, _ := json.MarshalIndent(result, "", " ")
|
||||
return mcp.NewToolResultText(string(output)), nil
|
||||
}
|
||||
}
|
||||
|
||||
func handleSearchKnowledge(store *qdrantStore) server.ToolHandlerFunc {
|
||||
func handleSearchKnowledge(sqlite *sqliteStore, qdrant *qdrantStore) server.ToolHandlerFunc {
|
||||
return func(ctx context.Context, request mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
||||
args, ok := request.Params.Arguments.(map[string]any)
|
||||
if !ok {
|
||||
@@ -177,12 +216,22 @@ func handleSearchKnowledge(store *qdrantStore) server.ToolHandlerFunc {
|
||||
}
|
||||
}
|
||||
|
||||
results, err := store.search(ctx, query, limit)
|
||||
results, err := qdrant.search(ctx, query, limit)
|
||||
if err != nil {
|
||||
return mcp.NewToolResultError(fmt.Sprintf("search failed: %v", err)), nil
|
||||
}
|
||||
|
||||
output, _ := json.MarshalIndent(results, "", " ")
|
||||
for i, r := range results {
|
||||
if cached, err := sqlite.getByID(ctx, r.Entry.ID); err == nil && cached != nil {
|
||||
results[i].Entry = *cached
|
||||
}
|
||||
}
|
||||
|
||||
output, err := json.MarshalIndent(results, "", " ")
|
||||
if err != nil {
|
||||
return mcp.NewToolResultError(fmt.Sprintf("failed to marshal results: %v", err)), nil
|
||||
}
|
||||
|
||||
return mcp.NewToolResultText(string(output)), nil
|
||||
}
|
||||
}
|
||||
@@ -195,8 +244,8 @@ func authMiddleware(token string, next http.Handler) http.HandlerFunc {
|
||||
return
|
||||
}
|
||||
|
||||
authHeader := r.Header.Get("Authorization")
|
||||
if !strings.HasPrefix(authHeader, "Bearer ") || strings.TrimPrefix(authHeader, "Bearer ") != token {
|
||||
authHeader := strings.ToLower(r.Header.Get("Authorization"))
|
||||
if !strings.HasPrefix(authHeader, "bearer ") || strings.TrimPrefix(authHeader, "bearer ") != token {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
w.Write([]byte(`{"error":"unauthorized"}`))
|
||||
@@ -222,10 +271,10 @@ func corsMiddleware(next http.Handler) http.HandlerFunc {
|
||||
}
|
||||
}
|
||||
|
||||
func requireEnv(key string) string {
|
||||
func requireEnv(logger *zap.Logger, key string) string {
|
||||
val := os.Getenv(key)
|
||||
if val == "" {
|
||||
log.Fatalf("environment variable %s is required", key)
|
||||
logger.Fatal("missing required env var", zap.String("key", key))
|
||||
}
|
||||
return val
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user