MCP Aggregator Component¶

Overview¶

The MCP Aggregator (internal/aggregator) is the core component responsible for unifying tools from multiple sources into a single, coherent interface for AI agents. It provides a unified MCP protocol interface that exposes:

11 Meta-Tools as the primary interface (list_tools, call_tool, etc.)
36 Core Built-in Tools across 5 functional categories (accessed via call_tool)
Dynamic Workflow Tools generated from workflow definitions (accessed via call_tool)
External Tools from configured MCP servers (accessed via call_tool)

The aggregator acts as an intelligent proxy that discovers, registers, filters, and routes tool calls to the appropriate underlying sources while providing consistent tool discovery and execution patterns.

Key Architectural Note: MCP clients see only the meta-tools. All actual tools (core_, workflow_, x_*) are accessed through the call_tool meta-tool. This design enables session-scoped tool visibility and unified authentication handling.

Meta-Tools Interface¶

The aggregator exposes only meta-tools as its MCP interface. This is the primary way clients interact with muster.

Tool Discovery Meta-Tools¶

Meta-Tool	Description
`list_tools`	List one bounded page of the current session's tools (`limit`/`offset`, `total`, `truncated`)
`describe_tool`	Get detailed schema for a specific tool
`filter_tools`	Search tools by pattern
`list_core_tools`	List only muster core tools

Tool Execution Meta-Tool¶

Meta-Tool	Description
`call_tool`	Execute any tool by name with arguments

Resource Meta-Tools¶

Meta-Tool	Description
`list_resources`	List available MCP resources
`describe_resource`	Get resource metadata
`get_resource`	Read resource contents

Prompt Meta-Tools¶

Meta-Tool	Description
`list_prompts`	List available prompts
`describe_prompt`	Get prompt details
`get_prompt`	Execute a prompt

Tool Execution Flow¶

All tool calls go through the call_tool meta-tool:

MCP Client → call_tool(name="core_workflow_list", args={})
           → Aggregator.CallToolInternal()
           → callCoreToolDirectly() or forward to backend MCP server
           → Result wrapped in structured JSON response

Session-Scoped Visibility¶

The list_tools response includes: 1. Available tools: One page (50 by default) of the tools from connected/authenticated servers, summarised; total and truncated tell the caller whether more pages exist 2. Servers requiring auth: Information about OAuth-protected servers that need authentication

Architecture¶

Core Responsibilities¶

Meta-Tools Interface: Expose meta-tools (list_tools, call_tool, etc.) as the only MCP interface
Core Tool Management: Provide built-in tools across 4 categories (config, mcpserver, service, workflow)
Dynamic Tool Generation: Generate workflow execution tools (workflow_*) from workflow definitions
MCP Server Discovery: Automatically discover and connect to configured external MCP servers
Tool Aggregation: Collect and register tools from all sources into a unified registry
Session-Scoped Visibility: Manage tool visibility based on authentication state
Request Routing: Route tool calls to appropriate handlers (core, workflow engine, or external servers)
Response Wrapping: Wrap tool responses in structured JSON for consistent handling

Component Structure¶

internal/aggregator/
├── server.go          # Main aggregator MCP server implementation
├── registry.go        # Tool registration and management
├── tool_factory.go    # Dynamic tool creation and proxying
├── event_handler.go   # Server lifecycle event processing
├── manager.go         # High-level orchestration
└── types.go          # Core data structures

Key Components¶

Registry (`registry.go`)¶

Purpose: Central tool registry managing all aggregated tools

Key Functionality: - Tool registration from multiple sources - Conflict resolution for duplicate tool names - Tool metadata management - Dynamic tool discovery

Implementation Pattern:

type Registry struct {
    tools     map[string]*ToolEntry
    servers   map[string]*ServerEntry
    mutex     sync.RWMutex
}

type ToolEntry struct {
    Name        string
    Description string
    Schema      *ToolSchema
    ServerID    string
    LastSeen    time.Time
}

Key Methods: - RegisterTool(serverID, toolName string, schema *ToolSchema) error - UnregisterTool(serverID, toolName string) error - GetTool(name string) (*ToolEntry, error) - ListTools(filter *ToolFilter) ([]*ToolEntry, error)

Tool Factory (`tool_factory.go`)¶

Purpose: Dynamic creation of meta-tools for tool discovery and execution

Meta-Tools Provided: 1. list_tools - Enumerate available tools with filtering 2. filter_tools - Apply filters to tool lists 3. call_tool - Execute tools on underlying servers 4. get_tool_schema - Retrieve detailed tool schemas

Implementation Pattern:

type ToolFactory struct {
    registry  *Registry
    mcpClient MCPClientInterface
}

func (f *ToolFactory) CreateListToolsTool() *Tool {
    return &Tool{
        Name:        "list_tools",
        Description: "List available tools from all connected MCP servers",
        Schema:      listToolsSchema,
        Handler:     f.handleListTools,
    }
}

func (f *ToolFactory) handleListTools(ctx context.Context, args map[string]interface{}) (interface{}, error) {
    filter := parseToolFilter(args)
    tools, err := f.registry.ListTools(filter)
    if err != nil {
        return nil, fmt.Errorf("failed to list tools: %w", err)
    }
    return formatToolList(tools), nil
}

Event Handler (`event_handler.go`)¶

Purpose: Process server lifecycle events and maintain tool registry consistency

Event Types: - Server connected/disconnected - Tool added/removed/updated - Server health status changes

Implementation Pattern:

type EventHandler struct {
    registry   *Registry
    notifier   EventNotifier
}

func (h *EventHandler) HandleServerConnected(serverID string, tools []*ToolDefinition) error {
    h.logger.Info("Server connected", "server_id", serverID, "tool_count", len(tools))

    for _, tool := range tools {
        if err := h.registry.RegisterTool(serverID, tool.Name, tool.Schema); err != nil {
            h.logger.Error("Failed to register tool", "tool", tool.Name, "error", err)
        }
    }

    h.notifier.NotifyServerStateChange(serverID, "connected")
    return nil
}

Integration Patterns¶

API Integration¶

The aggregator integrates with the central API through the adapter pattern:

// internal/aggregator/api_adapter.go
type Adapter struct {
    server *Server
    logger *slog.Logger
}

func (a *Adapter) ListTools(ctx context.Context, filter *ToolFilter) ([]*Tool, error) {
    return a.server.listTools(ctx, filter)
}

func (a *Adapter) CallTool(ctx context.Context, name string, args map[string]interface{}) (interface{}, error) {
    return a.server.callTool(ctx, name, args)
}

func (a *Adapter) Register() {
    api.RegisterAggregatorHandler(a)
}

MCP Server Communication¶

Communication with underlying MCP servers follows the standard MCP protocol:

type MCPClient interface {
    Connect(ctx context.Context, serverConfig *ServerConfig) error
    ListTools(ctx context.Context) ([]*ToolDefinition, error)
    CallTool(ctx context.Context, name string, args map[string]interface{}) (interface{}, error)
    Subscribe(handler EventHandler) error
    Disconnect(ctx context.Context) error
}

Usage Patterns¶

Tool Discovery Flow¶

Agent Request: AI agent calls list_tools meta-tool
Registry Query: Query tool registry with filters
Response Generation: Format and return tool list

func (s *Server) HandleListTools(ctx context.Context, request *ListToolsRequest) (*ListToolsResponse, error) {
    // Query registry
    tools, err := s.registry.ListTools(&ToolFilter{
        Patterns:    request.Patterns,
        ServerIDs:   request.ServerFilter,
        Categories:  request.Categories,
    })
    if err != nil {
        return nil, fmt.Errorf("failed to query tools: %w", err)
    }

    // Format response
    return &ListToolsResponse{
        Tools: formatToolDescriptions(tools),
        Total: len(tools),
    }, nil
}

Tool Execution Flow¶

Agent Request: AI agent calls call_tool with tool name and arguments
Tool Resolution: Resolve tool name to underlying server
Request Forwarding: Forward request to appropriate MCP server
Response Processing: Process and return response to agent

func (s *Server) HandleCallTool(ctx context.Context, request *CallToolRequest) (*CallToolResponse, error) {
    // Resolve tool to server
    tool, err := s.registry.GetTool(request.ToolName)
    if err != nil {
        return nil, fmt.Errorf("tool not found: %w", err)
    }

    // Get MCP client for server
    client, err := s.getServerClient(tool.ServerID)
    if err != nil {
        return nil, fmt.Errorf("server not available: %w", err)
    }

    // Forward request
    result, err := client.CallTool(ctx, request.ToolName, request.Arguments)
    if err != nil {
        return nil, fmt.Errorf("tool execution failed: %w", err)
    }

    return &CallToolResponse{
        Result: result,
        ToolName: request.ToolName,
        ServerID: tool.ServerID,
    }, nil
}

Configuration¶

Server Configuration¶

aggregator:
  # Server binding configuration
  bind:
    address: "127.0.0.1"
    port: 8080
    protocol: "stdio"  # or "http", "websocket"

  # Server discovery configuration
  discovery:
    auto_discovery: true
    discovery_interval: "30s"
    health_check_interval: "10s"

  # Performance tuning
  performance:
    max_concurrent_calls: 100
    tool_cache_ttl: "5m"
    connection_pool_size: 10

Error Handling¶

Error Categories¶

Connection Errors: MCP server unavailable or disconnected
Tool Errors: Tool not found or execution failure
Configuration Errors: Invalid configuration or missing dependencies
Authorization Errors: Caller not authorized for the requested tool

Error Response Format¶

type AggregatorError struct {
    Type      string `json:"type"`
    Message   string `json:"message"`
    ServerID  string `json:"server_id,omitempty"`
    ToolName  string `json:"tool_name,omitempty"`
    Timestamp string `json:"timestamp"`
}

func (e *AggregatorError) Error() string {
    return fmt.Sprintf("[%s] %s", e.Type, e.Message)
}

Performance Considerations¶

Caching Strategy¶

Tool Metadata: Cache tool schemas and descriptions
Server Status: Cache server health and availability
Response Caching: Cache responses for idempotent operations

Connection Management¶

Connection Pooling: Maintain persistent connections to MCP servers
Health Monitoring: Regular health checks for all connected servers
Graceful Degradation: Continue operation when servers are unavailable

Concurrency Handling¶

Request Concurrency: Handle multiple simultaneous tool calls
Registry Locking: Use read/write locks for registry access
Server Communication: Async communication with MCP servers

Monitoring and Observability¶

Metrics¶

Tool Call Rate: Number of tool calls per second
Server Health: Connected/disconnected server count
Error Rate: Failed tool calls and connection errors
Response Time: Tool call latency distribution

Logging¶

func (s *Server) logToolCall(ctx context.Context, toolName, serverID string, duration time.Duration, err error) {
    fields := []slog.Attr{
        slog.String("tool_name", toolName),
        slog.String("server_id", serverID),
        slog.Duration("duration", duration),
    }

    if err != nil {
        fields = append(fields, slog.String("error", err.Error()))
        s.logger.LogAttrs(ctx, slog.LevelError, "Tool call failed", fields...)
    } else {
        s.logger.LogAttrs(ctx, slog.LevelInfo, "Tool call completed", fields...)
    }
}

Testing Strategy¶

Unit Testing¶

func TestToolRegistration(t *testing.T) {
    registry := NewRegistry()

    tool := &ToolDefinition{
        Name:        "test_tool",
        Description: "Test tool",
        Schema:      &ToolSchema{},
    }

    err := registry.RegisterTool("server1", tool.Name, tool.Schema)
    assert.NoError(t, err)

    retrieved, err := registry.GetTool("test_tool")
    assert.NoError(t, err)
    assert.Equal(t, "test_tool", retrieved.Name)
    assert.Equal(t, "server1", retrieved.ServerID)
}

Integration Testing¶

func TestAggregatorIntegration(t *testing.T) {
    // Setup mock MCP servers
    mockServer1 := &MockMCPServer{
        Tools: []*ToolDefinition{
            {Name: "server1_tool", Description: "Tool from server 1"},
        },
    }

    // Create aggregator
    aggregator := NewAggregator()
    aggregator.AddServer("server1", mockServer1)

    // Test tool discovery
    tools, err := aggregator.ListTools(context.Background(), nil)
    assert.NoError(t, err)
    assert.Len(t, tools, 1)
    assert.Equal(t, "server1_tool", tools[0].Name)

    // Test tool execution
    result, err := aggregator.CallTool(context.Background(), "server1_tool", map[string]interface{}{})
    assert.NoError(t, err)
    assert.NotNil(t, result)
}

The MCP Aggregator serves as the intelligent hub that makes muster's unified tool interface possible, providing seamless integration between AI agents and multiple underlying MCP servers while maintaining security, performance, and reliability.

MCP Aggregator Component¶

Overview¶

Meta-Tools Interface¶

Tool Discovery Meta-Tools¶

Tool Execution Meta-Tool¶

Resource Meta-Tools¶

Prompt Meta-Tools¶

Tool Execution Flow¶

Session-Scoped Visibility¶

Architecture¶

Core Responsibilities¶

Component Structure¶

Key Components¶

Registry (registry.go)¶

Tool Factory (tool_factory.go)¶

Event Handler (event_handler.go)¶

Integration Patterns¶

API Integration¶

MCP Server Communication¶

Usage Patterns¶

Tool Discovery Flow¶

Tool Execution Flow¶

Configuration¶

Server Configuration¶

Error Handling¶

Error Categories¶

Error Response Format¶

Performance Considerations¶

Caching Strategy¶

Connection Management¶

Concurrency Handling¶

Monitoring and Observability¶

Metrics¶

Logging¶

Testing Strategy¶

Unit Testing¶

Integration Testing¶

Registry (`registry.go`)¶

Tool Factory (`tool_factory.go`)¶

Event Handler (`event_handler.go`)¶