server

package
v1.0.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jan 15, 2026 License: Apache-2.0 Imports: 39 Imported by: 0

Documentation

Overview

Copyright 2026 Teradata

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Index

Constants

View Source 
const (
	// DefaultProgressBufferSize is the default buffer size for progress event channels
	DefaultProgressBufferSize = 10
)

Variables

This section is empty.

Functions

func ConvertMessage 

func ConvertMessage(m *agent.Message) *loomv1.Message

ConvertMessage converts an agent.Message to proto format.

func ConvertSession 

func ConvertSession(s *agent.Session) *loomv1.Session

ConvertSession converts an agent.Session to proto format.

func ConvertTool 

func ConvertTool(t shuttle.Tool) *loomv1.ToolDefinition

ConvertTool converts a shuttle.Tool to proto format with rich metadata. Attempts to load metadata from YAML files; falls back to basic info if not found. Logs warnings if metadata loading fails but continues with basic tool definition.

func GenerateSessionID 

func GenerateSessionID() string

GenerateSessionID generates a new session ID.

Types

type CORSConfig added in v1.0.2 

type CORSConfig struct {
	Enabled          bool
	AllowedOrigins   []string
	AllowedMethods   []string
	AllowedHeaders   []string
	ExposedHeaders   []string
	AllowCredentials bool
	MaxAge           int
}

CORSConfig holds CORS configuration

func DefaultCORSConfig added in v1.0.2 

func DefaultCORSConfig() CORSConfig

DefaultCORSConfig returns a permissive CORS configuration

type HTTPServer 

type HTTPServer struct {
	// contains filtered or unexported fields
}

HTTPServer wraps gRPC server with HTTP/REST+SSE endpoints

func NewHTTPServer 

func NewHTTPServer(grpcServer *MultiAgentServer, httpAddr, grpcAddr string, logger *zap.Logger) *HTTPServer

NewHTTPServer creates an HTTP server that proxies to gRPC

func NewHTTPServerWithCORS added in v1.0.2 

func NewHTTPServerWithCORS(grpcServer *MultiAgentServer, httpAddr, grpcAddr string, logger *zap.Logger, corsConfig CORSConfig) *HTTPServer

NewHTTPServerWithCORS creates an HTTP server with custom CORS configuration

func (*HTTPServer) Start 

func (h *HTTPServer) Start(ctx context.Context) error

Start starts the HTTP server

func (*HTTPServer) Stop 

func (h *HTTPServer) Stop(ctx context.Context) error

Stop gracefully stops the HTTP server

type LearningService 

type LearningService struct {
	loomv1.UnimplementedLearningAgentServiceServer
	// contains filtered or unexported fields
}

LearningService implements the LearningAgentService gRPC interface. It wraps the LearningAgent and provides gRPC service endpoints.

func NewLearningService 

func NewLearningService(agent *learning.LearningAgent) *LearningService

NewLearningService creates a new LearningService gRPC wrapper

func (*LearningService) AnalyzePatternEffectiveness 

func (s *LearningService) AnalyzePatternEffectiveness(
	ctx context.Context,
	req *loomv1.AnalyzePatternEffectivenessRequest,
) (*loomv1.PatternAnalysisResponse, error)

AnalyzePatternEffectiveness analyzes pattern performance over a time window

func (*LearningService) ApplyImprovement 

func (s *LearningService) ApplyImprovement(
	ctx context.Context,
	req *loomv1.ApplyImprovementRequest,
) (*loomv1.ApplyImprovementResponse, error)

ApplyImprovement applies an improvement proposal

func (*LearningService) GenerateImprovements 

func (s *LearningService) GenerateImprovements(
	ctx context.Context,
	req *loomv1.GenerateImprovementsRequest,
) (*loomv1.ImprovementsResponse, error)

GenerateImprovements generates improvement proposals based on analysis

func (*LearningService) GetImprovementHistory 

func (s *LearningService) GetImprovementHistory(
	ctx context.Context,
	req *loomv1.GetImprovementHistoryRequest,
) (*loomv1.ImprovementHistoryResponse, error)

GetImprovementHistory retrieves improvement history

func (*LearningService) RollbackImprovement 

func (s *LearningService) RollbackImprovement(
	ctx context.Context,
	req *loomv1.RollbackImprovementRequest,
) (*loomv1.RollbackImprovementResponse, error)

RollbackImprovement rolls back a previously applied improvement

func (*LearningService) StreamPatternMetrics 

func (s *LearningService) StreamPatternMetrics(
	req *loomv1.StreamPatternMetricsRequest,
	stream loomv1.LearningAgentService_StreamPatternMetricsServer,
) error

StreamPatternMetrics streams real-time pattern metrics 🚨 CRITICAL: This implements the streaming RPC using MessageBus

type MultiAgentServer 

type MultiAgentServer struct {
	loomv1.UnimplementedLoomServiceServer
	// contains filtered or unexported fields
}

MultiAgentServer implements the LoomService gRPC server with support for multiple agents. It routes requests to the appropriate agent based on agent_id in the request.

func NewMultiAgentServer 

func NewMultiAgentServer(agents map[string]*agent.Agent, store *agent.SessionStore) *MultiAgentServer

NewMultiAgentServer creates a new multi-agent LoomService server.

func (*MultiAgentServer) AddAgent 

func (s *MultiAgentServer) AddAgent(id string, ag *agent.Agent)

AddAgent adds a new agent to the server at runtime

func (*MultiAgentServer) AddMCPServer 

func (s *MultiAgentServer) AddMCPServer(ctx context.Context, req *loomv1.AddMCPServerRequest) (*loomv1.AddMCPServerResponse, error)

AddMCPServer adds a new MCP server configuration.

func (*MultiAgentServer) AnswerClarificationQuestion 

func (s *MultiAgentServer) AnswerClarificationQuestion(ctx context.Context, req *loomv1.AnswerClarificationRequest) (*loomv1.AnswerClarificationResponse, error)

AnswerClarificationQuestion provides an answer to a clarification question asked by an agent.

func (*MultiAgentServer) ConfigureCommunication 

func (s *MultiAgentServer) ConfigureCommunication(
	bus *communication.MessageBus,
	queue *communication.MessageQueue,
	sharedMem *communication.SharedMemoryStore,
	refStore communication.ReferenceStore,
	policy *communication.PolicyManager,
	logger *zap.Logger,
) error

ConfigureCommunication initializes the tri-modal communication system for all agents. This should be called after NewMultiAgentServer() but before starting the server.

The three communication modes: 1. MessageBus - Broadcast/pub-sub for topic-based multicast 2. MessageQueue - Point-to-point async messaging with queuing 3. SharedMemoryStore - Zero-copy data sharing with namespaces

func (*MultiAgentServer) ConfigureLegacyCommunication 

func (s *MultiAgentServer) ConfigureLegacyCommunication(refStore communication.ReferenceStore, policy *communication.PolicyManager) error

ConfigureLegacyCommunication initializes basic inter-agent communication with reference store and policy. Deprecated: Use ConfigureCommunication() from communication_handlers.go for tri-modal communication.

func (*MultiAgentServer) ConfigureScheduler 

func (s *MultiAgentServer) ConfigureScheduler(sched *scheduler.Scheduler)

ConfigureScheduler injects the workflow scheduler for cron-based execution. This should be called after NewMultiAgentServer() to enable schedule management RPCs.

func (*MultiAgentServer) ConfigureSharedMemory 

func (s *MultiAgentServer) ConfigureSharedMemory(config *storage.Config) error

ConfigureSharedMemory initializes shared memory for all agents with the given configuration. This should be called after NewMultiAgentServer() but before starting the server. All agents will share the same memory store, enabling efficient data passing between them.

func (*MultiAgentServer) ConfigureTLS 

func (s *MultiAgentServer) ConfigureTLS(manager *tls.Manager, config *loomv1.ServerConfig)

ConfigureTLS sets the TLS manager and server configuration for this server. This should be called after NewMultiAgentServer() if TLS is enabled.

func (*MultiAgentServer) CreatePattern 

func (s *MultiAgentServer) CreatePattern(ctx context.Context, req *loomv1.CreatePatternRequest) (*loomv1.CreatePatternResponse, error)

CreatePattern creates a new pattern at runtime for a specific agent.

func (*MultiAgentServer) CreateSession 

func (s *MultiAgentServer) CreateSession(ctx context.Context, req *loomv1.CreateSessionRequest) (*loomv1.Session, error)

CreateSession creates a new conversation session for an agent.

func (*MultiAgentServer) DeleteArtifact 

func (s *MultiAgentServer) DeleteArtifact(ctx context.Context, req *loomv1.DeleteArtifactRequest) (*loomv1.DeleteArtifactResponse, error)

DeleteArtifact deletes an artifact (soft or hard delete).

func (*MultiAgentServer) DeleteMCPServer 

func (s *MultiAgentServer) DeleteMCPServer(ctx context.Context, req *loomv1.DeleteMCPServerRequest) (*loomv1.DeleteMCPServerResponse, error)

DeleteMCPServer removes an MCP server.

func (*MultiAgentServer) DeleteScheduledWorkflow 

func (s *MultiAgentServer) DeleteScheduledWorkflow(ctx context.Context, req *loomv1.DeleteScheduledWorkflowRequest) (*emptypb.Empty, error)

DeleteScheduledWorkflow deletes a scheduled workflow. YAML-sourced schedules cannot be deleted via RPC - they must be removed by deleting the YAML file.

func (*MultiAgentServer) DeleteSession 

func (s *MultiAgentServer) DeleteSession(ctx context.Context, req *loomv1.DeleteSessionRequest) (*loomv1.DeleteSessionResponse, error)

DeleteSession deletes a session.

func (*MultiAgentServer) DeleteSharedMemory 

func (s *MultiAgentServer) DeleteSharedMemory(ctx context.Context, req *loomv1.DeleteSharedMemoryRequest) (*loomv1.DeleteSharedMemoryResponse, error)

DeleteSharedMemory removes a value from shared memory.

func (*MultiAgentServer) ExecuteWorkflow 

func (s *MultiAgentServer) ExecuteWorkflow(ctx context.Context, req *loomv1.ExecuteWorkflowRequest) (*loomv1.ExecuteWorkflowResponse, error)

ExecuteWorkflow executes a workflow pattern loaded from YAML or programmatically defined. This RPC enables automatic execution of multi-agent workflows.

func (*MultiAgentServer) GetAgentIDs 

func (s *MultiAgentServer) GetAgentIDs() []string

GetAgentIDs returns available agent IDs (internal helper)

func (*MultiAgentServer) GetArtifact 

func (s *MultiAgentServer) GetArtifact(ctx context.Context, req *loomv1.GetArtifactRequest) (*loomv1.GetArtifactResponse, error)

GetArtifact retrieves artifact metadata.

func (*MultiAgentServer) GetArtifactContent 

func (s *MultiAgentServer) GetArtifactContent(ctx context.Context, req *loomv1.GetArtifactContentRequest) (*loomv1.GetArtifactContentResponse, error)

GetArtifactContent reads artifact file content.

func (*MultiAgentServer) GetArtifactStats 

func (s *MultiAgentServer) GetArtifactStats(ctx context.Context, req *loomv1.GetArtifactStatsRequest) (*loomv1.GetArtifactStatsResponse, error)

GetArtifactStats retrieves storage statistics.

func (*MultiAgentServer) GetCommunicationComponents 

func (s *MultiAgentServer) GetCommunicationComponents() (
	*communication.MessageBus,
	*communication.MessageQueue,
	*communication.SharedMemoryStore,
)

GetCommunicationComponents returns the communication system components (for testing/inspection).

func (*MultiAgentServer) GetCommunicationPolicy 

func (s *MultiAgentServer) GetCommunicationPolicy() *communication.PolicyManager

GetCommunicationPolicy returns the communication policy (for testing/inspection).

func (*MultiAgentServer) GetConversationHistory 

func (s *MultiAgentServer) GetConversationHistory(ctx context.Context, req *loomv1.GetConversationHistoryRequest) (*loomv1.ConversationHistory, error)

GetConversationHistory retrieves conversation history.

func (*MultiAgentServer) GetHealth 

func (s *MultiAgentServer) GetHealth(ctx context.Context, req *loomv1.GetHealthRequest) (*loomv1.HealthStatus, error)

GetHealth performs a health check.

func (*MultiAgentServer) GetMCPServer 

func (s *MultiAgentServer) GetMCPServer(ctx context.Context, req *loomv1.GetMCPServerRequest) (*loomv1.MCPServerInfo, error)

GetMCPServer retrieves a specific MCP server.

func (*MultiAgentServer) GetReferenceStore 

func (s *MultiAgentServer) GetReferenceStore() communication.ReferenceStore

GetReferenceStore returns the reference store (for testing/inspection).

func (*MultiAgentServer) GetScheduleHistory 

func (s *MultiAgentServer) GetScheduleHistory(ctx context.Context, req *loomv1.GetScheduleHistoryRequest) (*loomv1.GetScheduleHistoryResponse, error)

GetScheduleHistory returns execution history for a schedule. Includes the last N executions with their status, timing, and error information.

func (*MultiAgentServer) GetScheduledWorkflow 

func (s *MultiAgentServer) GetScheduledWorkflow(ctx context.Context, req *loomv1.GetScheduledWorkflowRequest) (*loomv1.ScheduledWorkflow, error)

GetScheduledWorkflow retrieves a scheduled workflow by ID.

func (*MultiAgentServer) GetServerConfig 

func (s *MultiAgentServer) GetServerConfig(ctx context.Context, req *loomv1.GetServerConfigRequest) (*loomv1.ServerConfig, error)

GetServerConfig returns the current server configuration.

func (*MultiAgentServer) GetSession 

func (s *MultiAgentServer) GetSession(ctx context.Context, req *loomv1.GetSessionRequest) (*loomv1.Session, error)

GetSession retrieves session details.

func (*MultiAgentServer) GetSharedMemory 

func (s *MultiAgentServer) GetSharedMemory(ctx context.Context, req *loomv1.GetSharedMemoryRequest) (*loomv1.GetSharedMemoryResponse, error)

GetSharedMemory retrieves a value from shared memory.

func (*MultiAgentServer) GetSharedMemoryStats 

func (s *MultiAgentServer) GetSharedMemoryStats(ctx context.Context, req *loomv1.GetSharedMemoryStatsRequest) (*loomv1.SharedMemoryStats, error)

GetSharedMemoryStats retrieves statistics for a namespace.

func (*MultiAgentServer) GetTLSStatus 

func (s *MultiAgentServer) GetTLSStatus(ctx context.Context, req *loomv1.GetTLSStatusRequest) (*loomv1.TLSStatus, error)

GetTLSStatus returns the current TLS/certificate status.

func (*MultiAgentServer) HealthCheckMCPServers 

func (s *MultiAgentServer) HealthCheckMCPServers(ctx context.Context, req *loomv1.HealthCheckMCPServersRequest) (*loomv1.HealthCheckMCPServersResponse, error)

HealthCheckMCPServers checks health of all MCP servers.

func (*MultiAgentServer) ListAgents 

func (s *MultiAgentServer) ListAgents(ctx context.Context, req *loomv1.ListAgentsRequest) (*loomv1.ListAgentsResponse, error)

ListAgents implements the gRPC ListAgents RPC method

func (*MultiAgentServer) ListArtifacts 

func (s *MultiAgentServer) ListArtifacts(ctx context.Context, req *loomv1.ListArtifactsRequest) (*loomv1.ListArtifactsResponse, error)

ListArtifacts lists artifacts with optional filtering.

func (*MultiAgentServer) ListAvailableModels 

func (s *MultiAgentServer) ListAvailableModels(ctx context.Context, req *loomv1.ListAvailableModelsRequest) (*loomv1.ListAvailableModelsResponse, error)

ListAvailableModels lists all LLM models available to this server instance. Uses the provider factory to dynamically determine which models are actually available based on configured credentials and environment variables.

func (*MultiAgentServer) ListMCPServerTools 

func (s *MultiAgentServer) ListMCPServerTools(ctx context.Context, req *loomv1.ListMCPServerToolsRequest) (*loomv1.ListMCPServerToolsResponse, error)

ListMCPServerTools lists all tools from a specific MCP server. This queries the MCP server directly through the manager, not the agent's tool registry.

func (*MultiAgentServer) ListMCPServers 

func (s *MultiAgentServer) ListMCPServers(ctx context.Context, req *loomv1.ListMCPServersRequest) (*loomv1.ListMCPServersResponse, error)

ListMCPServers lists all configured MCP servers.

func (*MultiAgentServer) ListScheduledWorkflows 

func (s *MultiAgentServer) ListScheduledWorkflows(ctx context.Context, req *loomv1.ListScheduledWorkflowsRequest) (*loomv1.ListScheduledWorkflowsResponse, error)

ListScheduledWorkflows lists all scheduled workflows. Can optionally filter to only enabled schedules.

func (*MultiAgentServer) ListSessions 

func (s *MultiAgentServer) ListSessions(ctx context.Context, req *loomv1.ListSessionsRequest) (*loomv1.ListSessionsResponse, error)

ListSessions lists all sessions across all agents.

func (*MultiAgentServer) ListSharedMemoryKeys 

func (s *MultiAgentServer) ListSharedMemoryKeys(ctx context.Context, req *loomv1.ListSharedMemoryKeysRequest) (*loomv1.ListSharedMemoryKeysResponse, error)

ListSharedMemoryKeys lists all keys matching a pattern in a namespace.

func (*MultiAgentServer) ListTools 

func (s *MultiAgentServer) ListTools(ctx context.Context, req *loomv1.ListToolsRequest) (*loomv1.ListToolsResponse, error)

ListTools lists all registered tools from the default agent. If req.Backend is specified, only tools for that backend are returned.

func (*MultiAgentServer) PauseSchedule 

func (s *MultiAgentServer) PauseSchedule(ctx context.Context, req *loomv1.PauseScheduleRequest) (*emptypb.Empty, error)

PauseSchedule pauses a schedule without deleting it. The schedule will remain in the database but will not execute until resumed.

func (*MultiAgentServer) Publish 

func (s *MultiAgentServer) Publish(ctx context.Context, req *loomv1.PublishRequest) (*loomv1.PublishResponse, error)

Publish publishes a message to a topic on the broadcast bus. All subscribers to the topic will receive the message.

func (*MultiAgentServer) PutSharedMemory 

func (s *MultiAgentServer) PutSharedMemory(ctx context.Context, req *loomv1.PutSharedMemoryRequest) (*loomv1.PutSharedMemoryResponse, error)

PutSharedMemory writes or updates a value in shared memory.

func (*MultiAgentServer) RegisterServerProgressListener 

func (s *MultiAgentServer) RegisterServerProgressListener(agentName string, logger *zap.Logger)

RegisterServerProgressListener creates and registers a progress listener for the server. This listener tracks pending clarification questions for TUI/RPC answer routing.

func (*MultiAgentServer) RemoveAgent 

func (s *MultiAgentServer) RemoveAgent(id string) error

RemoveAgent removes an agent from the server

func (*MultiAgentServer) RenewCertificate 

func (s *MultiAgentServer) RenewCertificate(ctx context.Context, req *loomv1.RenewCertificateRequest) (*loomv1.RenewCertificateResponse, error)

RenewCertificate manually triggers certificate renewal.

func (*MultiAgentServer) RestartMCPServer 

func (s *MultiAgentServer) RestartMCPServer(ctx context.Context, req *loomv1.RestartMCPServerRequest) (*loomv1.MCPServerInfo, error)

RestartMCPServer restarts a running MCP server.

func (*MultiAgentServer) ResumeSchedule 

func (s *MultiAgentServer) ResumeSchedule(ctx context.Context, req *loomv1.ResumeScheduleRequest) (*emptypb.Empty, error)

ResumeSchedule resumes a paused schedule. The schedule will start executing again according to its cron expression.

func (*MultiAgentServer) ScheduleWorkflow 

func (s *MultiAgentServer) ScheduleWorkflow(ctx context.Context, req *loomv1.ScheduleWorkflowRequest) (*loomv1.ScheduleWorkflowResponse, error)

ScheduleWorkflow creates a new scheduled workflow via RPC. The schedule will be persisted in SQLite and executed automatically by the cron engine.

func (*MultiAgentServer) SearchArtifacts 

func (s *MultiAgentServer) SearchArtifacts(ctx context.Context, req *loomv1.SearchArtifactsRequest) (*loomv1.SearchArtifactsResponse, error)

SearchArtifacts performs full-text search on artifacts.

func (*MultiAgentServer) SendAndReceive 

func (s *MultiAgentServer) SendAndReceive(ctx context.Context, req *loomv1.SendAndReceiveRequest) (*loomv1.SendAndReceiveResponse, error)

SendAndReceive sends a request message and waits for a response (RPC-style). This implements synchronous request-response communication via the message queue.

func (*MultiAgentServer) SendAsync 

func (s *MultiAgentServer) SendAsync(ctx context.Context, req *loomv1.SendAsyncRequest) (*loomv1.SendAsyncResponse, error)

SendAsync sends a fire-and-forget message to another agent via the message queue. If the destination agent is offline, the message is queued for later delivery.

func (*MultiAgentServer) SetAgentRegistry 

func (s *MultiAgentServer) SetAgentRegistry(registry *agent.Registry)

SetAgentRegistry injects the agent registry for workflow execution. This should be called after NewMultiAgentServer() to enable ExecuteWorkflow RPC.

func (*MultiAgentServer) SetArtifactStore 

func (s *MultiAgentServer) SetArtifactStore(store artifacts.ArtifactStore)

SetArtifactStore sets the artifact store for file management. This should be called after NewMultiAgentServer() to enable artifact operations.

func (*MultiAgentServer) SetClarificationConfig 

func (s *MultiAgentServer) SetClarificationConfig(channelSendTimeoutMs int)

SetClarificationConfig sets the clarification question timeout configuration.

func (*MultiAgentServer) SetLLMConcurrencyLimit 

func (s *MultiAgentServer) SetLLMConcurrencyLimit(limit int)

SetLLMConcurrencyLimit configures the maximum number of concurrent LLM calls. This prevents rate limiting by serializing or limiting parallel LLM API calls.

limit = 1: Fully serialized (safest for strict rate limits) limit = 2-5: Some parallelism while staying under quota limit > 5: Higher throughput, higher risk of rate limiting

This should be called after NewMultiAgentServer() but before spawning workflow sub-agents.

func (*MultiAgentServer) SetLogger 

func (s *MultiAgentServer) SetLogger(logger *zap.Logger)

SetLogger injects the logger for server operations. This should be called after NewMultiAgentServer() to enable logging.

func (*MultiAgentServer) SetMCPManager 

func (s *MultiAgentServer) SetMCPManager(mgr *manager.Manager, configPath string, logger *zap.Logger)

SetMCPManager injects the MCP manager for runtime management. This should be called after NewMultiAgentServer() to enable MCP server management.

func (*MultiAgentServer) SetProgressMultiplexer 

func (s *MultiAgentServer) SetProgressMultiplexer(agentID string, pm *metaagent.ProgressMultiplexer)

SetProgressMultiplexer sets the progress multiplexer for an agent. This enables multi-turn conversations and progress event broadcasting.

func (*MultiAgentServer) SetProviderFactory 

func (s *MultiAgentServer) SetProviderFactory(f *factory.ProviderFactory)

SetProviderFactory sets the LLM provider factory for dynamic model switching.

func (*MultiAgentServer) SetToolRegistry 

func (s *MultiAgentServer) SetToolRegistry(registry *toolregistry.Registry)

SetToolRegistry injects the tool registry for dynamic tool discovery. This should be called after NewMultiAgentServer() to enable tool_search indexing of MCP tools.

func (*MultiAgentServer) SharedMemoryStore 

func (s *MultiAgentServer) SharedMemoryStore() *storage.SharedMemoryStore

GetSharedMemory returns the shared memory store (for testing/inspection).

func (*MultiAgentServer) Shutdown 

func (s *MultiAgentServer) Shutdown(ctx context.Context) error

Shutdown gracefully shuts down the server, closing all pending question channels. This should be called during server shutdown to notify waiting agents that no more answers will be received.

func (*MultiAgentServer) StartHotReload 

func (s *MultiAgentServer) StartHotReload(ctx context.Context, logger *zap.Logger) error

StartHotReload initializes hot-reload watchers for all agents with pattern directories. Hot-reload events are automatically broadcast to clients via StreamPatternUpdates.

func (*MultiAgentServer) StartMessageQueueMonitor 

func (s *MultiAgentServer) StartMessageQueueMonitor(ctx context.Context)

StartMessageQueueMonitor starts a background goroutine that monitors the message queue and notifies workflow sub-agents when they have pending messages (event-driven, not polling).

func (*MultiAgentServer) StopHotReload 

func (s *MultiAgentServer) StopHotReload() error

StopHotReload stops all hot-reload watchers. Should be called during server shutdown.

func (*MultiAgentServer) StreamPatternUpdates 

func (s *MultiAgentServer) StreamPatternUpdates(req *loomv1.StreamPatternUpdatesRequest, stream loomv1.LoomService_StreamPatternUpdatesServer) error

StreamPatternUpdates streams pattern update events to clients in real-time. Used by TUI and other clients to show live pattern changes without polling.

func (*MultiAgentServer) StreamWeave 

func (s *MultiAgentServer) StreamWeave(req *loomv1.WeaveRequest, stream loomv1.LoomService_StreamWeaveServer) error

StreamWeave streams agent execution progress.

func (*MultiAgentServer) StreamWorkflow 

func (s *MultiAgentServer) StreamWorkflow(req *loomv1.ExecuteWorkflowRequest, stream loomv1.LoomService_StreamWorkflowServer) error

StreamWorkflow executes a workflow and streams progress updates to the client. This provides real-time feedback during long-running multi-agent workflows.

func (*MultiAgentServer) Subscribe 

func (s *MultiAgentServer) Subscribe(req *loomv1.SubscribeRequest, stream loomv1.LoomService_SubscribeServer) error

Subscribe creates a subscription to a topic and streams messages back to the client. The stream will remain open until the client closes it or an error occurs.

func (*MultiAgentServer) SubscribeToSession 

func (s *MultiAgentServer) SubscribeToSession(req *loomv1.SubscribeToSessionRequest, stream loomv1.LoomService_SubscribeToSessionServer) error

SubscribeToSession subscribes to real-time updates for a session. Streams updates when new messages arrive in the session conversation. This allows clients to receive asynchronous responses from workflow coordinators and sub-agents without polling.

func (*MultiAgentServer) SwitchModel 

func (s *MultiAgentServer) SwitchModel(ctx context.Context, req *loomv1.SwitchModelRequest) (*loomv1.SwitchModelResponse, error)

SwitchModel switches the LLM provider/model for a specific agent. The agent is identified by agent_id in the request.

func (*MultiAgentServer) TestMCPServerConnection 

func (s *MultiAgentServer) TestMCPServerConnection(ctx context.Context, req *loomv1.TestMCPServerConnectionRequest) (*loomv1.TestMCPServerConnectionResponse, error)

TestMCPServerConnection tests an MCP server configuration without persisting it. This allows users to validate their configuration before saving.

func (*MultiAgentServer) TriggerScheduledWorkflow 

func (s *MultiAgentServer) TriggerScheduledWorkflow(ctx context.Context, req *loomv1.TriggerScheduledWorkflowRequest) (*loomv1.ExecuteWorkflowResponse, error)

TriggerScheduledWorkflow manually triggers a scheduled workflow immediately. This bypasses the cron schedule and executes the workflow right away.

func (*MultiAgentServer) UpdateAgent 

func (s *MultiAgentServer) UpdateAgent(id string, ag *agent.Agent) error

UpdateAgent replaces an existing agent with a new instance (for hot-reload). The new agent will inherit shared memory and communication configuration if set.

func (*MultiAgentServer) UpdateMCPServer 

func (s *MultiAgentServer) UpdateMCPServer(ctx context.Context, req *loomv1.UpdateMCPServerRequest) (*loomv1.MCPServerInfo, error)

UpdateMCPServer updates an existing MCP server configuration.

func (*MultiAgentServer) UpdateScheduledWorkflow 

func (s *MultiAgentServer) UpdateScheduledWorkflow(ctx context.Context, req *loomv1.UpdateScheduledWorkflowRequest) (*loomv1.ScheduleWorkflowResponse, error)

UpdateScheduledWorkflow updates an existing scheduled workflow. YAML-sourced schedules cannot be updated via RPC - they must be modified by editing the YAML file.

func (*MultiAgentServer) UploadArtifact 

func (s *MultiAgentServer) UploadArtifact(ctx context.Context, req *loomv1.UploadArtifactRequest) (*loomv1.UploadArtifactResponse, error)

UploadArtifact uploads a file to artifacts storage.

func (*MultiAgentServer) WatchSharedMemory 

func (s *MultiAgentServer) WatchSharedMemory(req *loomv1.WatchSharedMemoryRequest, stream loomv1.LoomService_WatchSharedMemoryServer) error

WatchSharedMemory watches for changes to keys in a namespace and streams updates.

func (*MultiAgentServer) Weave 

func (s *MultiAgentServer) Weave(ctx context.Context, req *loomv1.WeaveRequest) (*loomv1.WeaveResponse, error)

Weave executes a user query using the specified agent.

type PatternEventBroadcaster 

type PatternEventBroadcaster struct {
	// contains filtered or unexported fields
}

PatternEventBroadcaster broadcasts pattern update events to multiple subscribers. Thread-safe for concurrent subscribe/unsubscribe/broadcast operations.

func NewPatternEventBroadcaster 

func NewPatternEventBroadcaster() *PatternEventBroadcaster

NewPatternEventBroadcaster creates a new pattern event broadcaster.

func (*PatternEventBroadcaster) Broadcast 

func (b *PatternEventBroadcaster) Broadcast(event *loomv1.PatternUpdateEvent)

Broadcast sends an event to all subscribers. Non-blocking: if a subscriber's channel is full, the event is dropped for that subscriber.

func (*PatternEventBroadcaster) BroadcastPatternCreated 

func (b *PatternEventBroadcaster) BroadcastPatternCreated(agentID, patternName, category, filePath string)

BroadcastPatternCreated broadcasts a pattern creation event.

func (*PatternEventBroadcaster) BroadcastPatternDeleted 

func (b *PatternEventBroadcaster) BroadcastPatternDeleted(agentID, patternName, category string)

BroadcastPatternDeleted broadcasts a pattern deletion event.

func (*PatternEventBroadcaster) BroadcastPatternModified 

func (b *PatternEventBroadcaster) BroadcastPatternModified(agentID, patternName, category, filePath string)

BroadcastPatternModified broadcasts a pattern modification event.

func (*PatternEventBroadcaster) BroadcastPatternValidationFailed 

func (b *PatternEventBroadcaster) BroadcastPatternValidationFailed(agentID, patternName, errorMsg string)

BroadcastPatternValidationFailed broadcasts a pattern validation failure event.

func (*PatternEventBroadcaster) Close 

func (b *PatternEventBroadcaster) Close()

Close closes all subscriber channels.

func (*PatternEventBroadcaster) Subscribe 

func (b *PatternEventBroadcaster) Subscribe() chan *loomv1.PatternUpdateEvent

Subscribe registers a new subscriber and returns a channel for receiving events. The channel buffer size is 100 to handle bursts of events. Caller must call Unsubscribe() to clean up.

func (*PatternEventBroadcaster) Unsubscribe 

func (b *PatternEventBroadcaster) Unsubscribe(ch chan *loomv1.PatternUpdateEvent)

Unsubscribe removes a subscriber and closes their channel.

type Server 

type Server struct {
	loomv1.UnimplementedLoomServiceServer
	// contains filtered or unexported fields
}

Server implements the LoomService gRPC server.

func NewServer 

func NewServer(ag *agent.Agent, store *agent.SessionStore) *Server

NewServer creates a new LoomService server.

func (*Server) CreateSession 

func (s *Server) CreateSession(ctx context.Context, req *loomv1.CreateSessionRequest) (*loomv1.Session, error)

CreateSession creates a new conversation session.

func (*Server) DeleteSession 

func (s *Server) DeleteSession(ctx context.Context, req *loomv1.DeleteSessionRequest) (*loomv1.DeleteSessionResponse, error)

DeleteSession deletes a session.

func (*Server) GetConversationHistory 

func (s *Server) GetConversationHistory(ctx context.Context, req *loomv1.GetConversationHistoryRequest) (*loomv1.ConversationHistory, error)

GetConversationHistory retrieves conversation history.

func (*Server) GetHealth 

func (s *Server) GetHealth(ctx context.Context, req *loomv1.GetHealthRequest) (*loomv1.HealthStatus, error)

GetHealth performs a health check.

func (*Server) GetPattern 

func (s *Server) GetPattern(ctx context.Context, req *loomv1.GetPatternRequest) (*loomv1.Pattern, error)

GetPattern retrieves a specific pattern.

func (*Server) GetSession 

func (s *Server) GetSession(ctx context.Context, req *loomv1.GetSessionRequest) (*loomv1.Session, error)

GetSession retrieves session details.

func (*Server) GetTrace 

func (s *Server) GetTrace(ctx context.Context, req *loomv1.GetTraceRequest) (*loomv1.Trace, error)

GetTrace retrieves execution trace.

func (*Server) ListAvailableModels 

func (s *Server) ListAvailableModels(ctx context.Context, req *loomv1.ListAvailableModelsRequest) (*loomv1.ListAvailableModelsResponse, error)

ListAvailableModels lists all available LLM models/providers.

func (*Server) ListAvailableModelsLegacy 

func (s *Server) ListAvailableModelsLegacy(ctx context.Context, req *loomv1.ListAvailableModelsRequest) (*loomv1.ListAvailableModelsResponse, error)

ListAvailableModelsLegacy is the old static implementation (deprecated).

func (*Server) ListPatterns 

func (s *Server) ListPatterns(ctx context.Context, req *loomv1.ListPatternsRequest) (*loomv1.ListPatternsResponse, error)

ListPatterns lists available patterns.

func (*Server) ListSessions 

func (s *Server) ListSessions(ctx context.Context, req *loomv1.ListSessionsRequest) (*loomv1.ListSessionsResponse, error)

ListSessions lists all sessions.

func (*Server) ListTools 

func (s *Server) ListTools(ctx context.Context, req *loomv1.ListToolsRequest) (*loomv1.ListToolsResponse, error)

ListTools lists all registered tools. If req.Backend is specified, only tools for that backend are returned.

func (*Server) LoadPatterns 

func (s *Server) LoadPatterns(ctx context.Context, req *loomv1.LoadPatternsRequest) (*loomv1.LoadPatternsResponse, error)

LoadPatterns loads pattern definitions.

func (*Server) RegisterTool 

func (s *Server) RegisterTool(ctx context.Context, req *loomv1.RegisterToolRequest) (*loomv1.RegisterToolResponse, error)

RegisterTool dynamically registers a new tool.

func (*Server) RequestToolPermission 

func (s *Server) RequestToolPermission(ctx context.Context, req *loomv1.ToolPermissionRequest) (*loomv1.ToolPermissionResponse, error)

RequestToolPermission requests user permission to execute a tool.

func (*Server) SetProviderFactory 

func (s *Server) SetProviderFactory(f *factory.ProviderFactory)

SetProviderFactory sets the LLM provider factory for dynamic model switching.

func (*Server) StreamWeave 

func (s *Server) StreamWeave(req *loomv1.WeaveRequest, stream loomv1.LoomService_StreamWeaveServer) error

StreamWeave streams agent execution progress.

func (*Server) SwitchModel 

func (s *Server) SwitchModel(ctx context.Context, req *loomv1.SwitchModelRequest) (*loomv1.SwitchModelResponse, error)

SwitchModel switches the LLM model/provider for a session.

func (*Server) Weave 

func (s *Server) Weave(ctx context.Context, req *loomv1.WeaveRequest) (*loomv1.WeaveResponse, error)

Weave executes a user query using the agent.

type WorkflowExecution 

type WorkflowExecution struct {
	ExecutionID string
	Pattern     *loomv1.WorkflowPattern
	Result      *loomv1.WorkflowResult
	StartTime   time.Time
	EndTime     time.Time
	Status      WorkflowStatus
	Error       string // Error message if failed
}

WorkflowExecution represents a single workflow execution with its metadata and results.

type WorkflowStatus 

type WorkflowStatus string

WorkflowStatus represents the execution state. 

const (
	WorkflowStatusRunning   WorkflowStatus = "running"
	WorkflowStatusCompleted WorkflowStatus = "completed"
	WorkflowStatusFailed    WorkflowStatus = "failed"
	WorkflowStatusCanceled  WorkflowStatus = "canceled"
)

type WorkflowStore 

type WorkflowStore struct {
	// contains filtered or unexported fields
}

WorkflowStore persists workflow execution history in memory. In production, this could be backed by a database.

func NewWorkflowStore 

func NewWorkflowStore() *WorkflowStore

NewWorkflowStore creates a new workflow execution store.

func (*WorkflowStore) Count 

func (s *WorkflowStore) Count() int

Count returns the total number of workflow executions.

func (*WorkflowStore) Delete 

func (s *WorkflowStore) Delete(executionID string) error

Delete removes a workflow execution from the store.

func (*WorkflowStore) Get 

func (s *WorkflowStore) Get(executionID string) (*WorkflowExecution, error)

Get retrieves a workflow execution by ID.

func (*WorkflowStore) List 

func (s *WorkflowStore) List(status WorkflowStatus) []*WorkflowExecution

List returns all workflow executions, optionally filtered by status.

func (*WorkflowStore) Store 

func (s *WorkflowStore) Store(exec *WorkflowExecution)

Store persists a workflow execution record.

func (*WorkflowStore) StoreResult 

func (s *WorkflowStore) StoreResult(executionID string, result *loomv1.WorkflowResult) error

StoreResult updates the result of a completed workflow execution.

func (*WorkflowStore) UpdateStatus 

func (s *WorkflowStore) UpdateStatus(executionID string, status WorkflowStatus, errorMsg string) error

UpdateStatus updates the status of a workflow execution.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.