mirror of
https://github.com/memohai/Memoh.git
synced 2026-04-27 07:16:19 +09:00
feat(mcp): replace bind-mount+exec with in-container gRPC service (#179)
Replace the host bind-mount + containerd exec approach with a per-bot
in-container gRPC server (ContainerService, port 9090). All file I/O,
exec, and MCP stdio sessions now go through gRPC instead of running
shell commands or reading host-mounted directories.
Architecture changes:
- cmd/mcp: rewritten as a gRPC server (ContainerService) with full
file and exec API (ReadFile, WriteFile, ListDir, ReadRaw, WriteRaw,
Exec, Stat, Mkdir, Rename, DeleteFile)
- internal/mcp/mcpcontainer: protobuf definitions and generated stubs
- internal/mcp/mcpclient: gRPC client wrapper with connection pool
(Pool) and Provider interface for dependency injection
- mcp.Manager: add per-bot IP cache, gRPC connection pool, and
SetContainerIP/MCPClient methods; remove DataDir/Exec helpers
- containerd.Service: remove ExecTask/ExecTaskStreaming; network setup
now returns NetworkResult{IP} for pool routing
- internal/fs/service.go: deleted (replaced by mcpclient)
- handlers/fs.go: deleted; MCP stdio session logic moved to mcp_stdio.go
- container provider Executor: all tools (read/write/list/edit/exec)
now call gRPC client instead of running shell via exec
- storefs, containerfs, media, skills, memory: all I/O ported to
mcpclient.Provider
Database:
- migration 0022: drop host_path column from containers table
One-time data migration:
- migrateBindMountData: on first Start() after upgrade, copies old
bind-mount data into the container via gRPC, then renames src dir
to prevent re-migration; runs in background goroutine
Bug fixes:
- mcp_stdio: callRaw now returns full JSON-RPC envelope
{"jsonrpc","id","result"|"error"} matching protocol spec;
explicit "initialize" call now advances session init state to
prevent duplicate handshake on next non-initialize call
- mcpclient Pool: properly evict stale gRPC connection after snapshot
replace (container process recreated); use SetContainerIP instead
of direct map write so IP changes always evict pool entry
- migrateBindMountData: walkErr on directories now counted as failure
so partially-walked trees don't get incorrectly marked as migrated
- cmd/mcp/Dockerfile: removed dead file (docker/Dockerfile.mcp is the
canonical production build)
Tests:
- provider_test.go: restored with bufconn in-process gRPC mock
(fakeContainerService + staticProvider), 14 cases covering all 5
tools plus edge cases
- mcp_session_test.go: new, covers JSON-RPC envelope, init state
machine, pending cleanup on cancel/close, readLoop cancel
- storefs/service_test.go: restored (pure function roundtrip tests)
This commit is contained in:
+11
-14
@@ -220,9 +220,9 @@ func provideAgentRuntimeManager(log *slog.Logger, cfg config.Config) *agentrunti
|
||||
func provideMemoryLLM(modelsService *models.Service, queries *dbsqlc.Queries, log *slog.Logger) memprovider.LLM {
|
||||
return &lazyLLMClient{modelsService: modelsService, queries: queries, timeout: 30 * time.Second, logger: log}
|
||||
}
|
||||
func provideMemoryProviderRegistry(log *slog.Logger, chatService *conversation.Service, accountService *accounts.Service, containerdHandler *handlers.ContainerdHandler) *memprovider.Registry {
|
||||
func provideMemoryProviderRegistry(log *slog.Logger, chatService *conversation.Service, accountService *accounts.Service, manager *mcp.Manager) *memprovider.Registry {
|
||||
registry := memprovider.NewRegistry(log)
|
||||
builtinRuntime := handlers.NewBuiltinMemoryRuntime(containerdHandler.FSService())
|
||||
builtinRuntime := handlers.NewBuiltinMemoryRuntime(manager)
|
||||
registry.RegisterFactory(memprovider.BuiltinType, func(id string, config map[string]any) (memprovider.Provider, error) {
|
||||
return memprovider.NewBuiltinProvider(log, builtinRuntime, chatService, accountService), nil
|
||||
})
|
||||
@@ -335,7 +335,7 @@ func provideMemoryHandler(log *slog.Logger, botService *bots.Service, accountSer
|
||||
h := handlers.NewMemoryHandler(log, botService, accountService)
|
||||
h.SetMemoryRegistry(memoryRegistry)
|
||||
h.SetSettingsService(settingsService)
|
||||
h.SetFSService(containerdHandler.FSService())
|
||||
h.SetMCPClientProvider(manager)
|
||||
return h
|
||||
}
|
||||
func provideAuthHandler(log *slog.Logger, accountService *accounts.Service, rc *boot.RuntimeConfig) *handlers.AuthHandler {
|
||||
@@ -356,16 +356,9 @@ func (h *memohAuthHandler) Register(e *echo.Echo) {
|
||||
e.POST("/api/auth/login", h.inner.Login)
|
||||
e.POST("/api/auth/refresh", h.inner.Refresh)
|
||||
}
|
||||
func provideMediaService(log *slog.Logger, cfg config.Config) (*media.Service, error) {
|
||||
dataRoot := strings.TrimSpace(cfg.MCP.DataRoot)
|
||||
if dataRoot == "" {
|
||||
dataRoot = config.DefaultDataRoot
|
||||
}
|
||||
provider, err := containerfs.New(dataRoot)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("init media provider: %w", err)
|
||||
}
|
||||
return media.NewService(log, provider), nil
|
||||
func provideMediaService(log *slog.Logger, manager *mcp.Manager) *media.Service {
|
||||
provider := containerfs.New(manager)
|
||||
return media.NewService(log, provider)
|
||||
}
|
||||
func provideUsersHandler(log *slog.Logger, accountService *accounts.Service, identityService *identities.Service, botService *bots.Service, routeService *route.DBService, channelStore *channel.Store, channelLifecycle *channel.Lifecycle, channelManager *channel.Manager, registry *channel.Registry) *handlers.UsersHandler {
|
||||
return handlers.NewUsersHandler(log, accountService, identityService, botService, routeService, channelStore, channelLifecycle, channelManager, registry)
|
||||
@@ -506,7 +499,7 @@ func startAgentRuntime(lc fx.Lifecycle, manager *agentruntime.Manager) {
|
||||
OnStop: func(ctx context.Context) error { return manager.Stop(ctx) },
|
||||
})
|
||||
}
|
||||
func startServer(lc fx.Lifecycle, logger *slog.Logger, srv *memohServer, shutdowner fx.Shutdowner, cfg config.Config, queries *dbsqlc.Queries, botService *bots.Service, containerdHandler *handlers.ContainerdHandler, mcpConnService *mcp.ConnectionService, toolGateway *mcp.ToolGatewayService, channelManager *channel.Manager) {
|
||||
func startServer(lc fx.Lifecycle, logger *slog.Logger, srv *memohServer, shutdowner fx.Shutdowner, cfg config.Config, queries *dbsqlc.Queries, botService *bots.Service, containerdHandler *handlers.ContainerdHandler, manager *mcp.Manager, mcpConnService *mcp.ConnectionService, toolGateway *mcp.ToolGatewayService, channelManager *channel.Manager) {
|
||||
fmt.Printf("Starting Memoh Agent %s\n", version.GetInfo())
|
||||
lc.Append(fx.Hook{
|
||||
OnStart: func(ctx context.Context) error {
|
||||
@@ -514,6 +507,10 @@ func startServer(lc fx.Lifecycle, logger *slog.Logger, srv *memohServer, shutdow
|
||||
return err
|
||||
}
|
||||
botService.SetContainerLifecycle(containerdHandler)
|
||||
botService.SetContainerReachability(func(ctx context.Context, botID string) error {
|
||||
_, err := manager.MCPClient(ctx, botID)
|
||||
return err
|
||||
})
|
||||
botService.AddRuntimeChecker(healthcheck.NewRuntimeCheckerAdapter(mcpchecker.NewChecker(logger, mcpConnService, toolGateway)))
|
||||
botService.AddRuntimeChecker(healthcheck.NewRuntimeCheckerAdapter(channelchecker.NewChecker(logger, channelManager)))
|
||||
go func() {
|
||||
|
||||
Reference in New Issue
Block a user