mcpmux

A Go library for building MCP (Model Context Protocol) servers.

Overview

mcpmux provides a clean, idiomatic Go API for implementing MCP servers that can expose tools, resources, and prompts to LLM applications.

Why?

Kirha is a premium data layer for agents. We run a cluster of MCP servers that expose data from our partner providers. We handle routing and micropayment. Official MCP SDKs don't support dynamic tool registration or runtime updates. Every MCP update meant rebuilding our entire MCP gateway. We fixed that, following community standards, so that you can too.

Features

• ✅ Full MCP 2025-03-26 Support: Complete implementation of the Model Context Protocol specification
• ✅ Multiple Transports: Stdio and HTTP/SSE transport support
• ✅ Tools Management: Easy tools registration, update and execution
• ✅ Prompt Templates: Dynamic prompt generation with argument support
• ✅ Middleware Support: Extensible middleware system over standard HTTP for HTTP transport. Add any of your existing HTTP middleware as-is (e.g., authentication, usage tracking, CORS)
• ✅ Production Ready: Thread-safe, well-tested, and designed for production use
• ✅ Gorilla/mux Style: Familiar API for Go developers
• ✅ Minimal Dependencies: Only requires golang.org/x/sync for advanced concurrency

Installation

go get go.kirha.ai/mcpmux

Quick Start

Basic Server with Tools

package main

import (
    "context"
    "errors"
    "log"

    "go.kirha.ai/mcpmux"
)

func main() {
    // Create server
    server := mcpmux.NewServer(
        mcpmux.WithServerInfo("my-app", "1.0.0"),
        mcpmux.WithInstructions("A simple MCP server example"),
        mcpmux.WithLogging(),
    )

    // Create tool aggregator
    tools := mcpmux.NewStaticToolAggregator()

    // Register a simple tool
    tools.Register("greet", greetTool,
        mcpmux.WithDescription("Greets a person by name"),
        mcpmux.WithParameters(
            mcpmux.StringParam("name", "Name of the person to greet", true),
        ))

    // Attach tools to server
    server.UseToolAggregator(tools)

    // Create HTTP transport (default localhost:8080)
    transport := mcpmux.NewHTTPTransport(nil)

    // Start server
    log.Fatal(server.Serve(transport))
}

func greetTool(ctx context.Context, args map[string]interface{}) (*mcpmux.ToolResult, error) {
    name, ok := args["name"].(string)
    if !ok {
        return mcpmux.ErrorResult(errors.New("name parameter is required")), nil
    }
    
    return mcpmux.TextResult("Hello, " + name + "!"), nil
}

Complete Example with All Features

package main

import (
    "context"
    "errors"
    "fmt"
    "log"

    "go.kirha.ai/mcpmux"
)

func main() {
    // Create server with capabilities
    server := mcpmux.NewServer(
        mcpmux.WithServerInfo("full-featured-server", "1.0.0"),
        mcpmux.WithInstructions("Complete MCP server example"),
        mcpmux.WithLogging(),
    )

    // Setup tools
    setupTools(server)
    
    // Setup resources
    setupResources(server)
    
    // Setup prompts
    setupPrompts(server)

    // Add MCP middleware (for all transports)
    server.Use(loggingMiddleware)

    // Create HTTP transport with HTTP-specific middleware
    transport := mcpmux.NewHTTPTransport(&mcpmux.HTTPTransportConfig{
        Addr: ":8080",
        Middleware: []func(http.Handler) http.Handler{
            mcpmux.RecoveryMiddleware(log.Default()),
            mcpmux.LoggingMiddleware(log.Default()),
            mcpmux.CORSMiddleware([]string{"*"}),
        },
    })

    log.Println("Starting MCP server on :8080")
    log.Fatal(server.Serve(transport))
}

func setupTools(server *mcpmux.Server) {
    tools := mcpmux.NewStaticToolAggregator()
    
    // Calculator tool
    tools.Register("calculate", func(ctx context.Context, args map[string]interface{}) (*mcpmux.ToolResult, error) {
        expression, ok := args["expression"].(string)
        if !ok {
            return mcpmux.ErrorResult(errors.New("expression required")), nil
        }
        
        // Simple calculator implementation
        result := fmt.Sprintf("Result: %s", expression) // Simplified for example
        return mcpmux.TextResult(result), nil
    },
        mcpmux.WithDescription("Evaluates mathematical expressions"),
        mcpmux.WithParameters(
            mcpmux.StringParam("expression", "Mathematical expression to evaluate", true),
        ))
    
    server.UseToolAggregator(tools)
}

func setupResources(server *mcpmux.Server) {
    resources := mcpmux.NewStaticResourceProvider()
    
    // Configuration resource
    resources.Register("config://settings", func(ctx context.Context, uri string) ([]mcpmux.ResourceContents, error) {
        config := `{"name": "my-app", "version": "1.0.0"}`
        return mcpmux.JSONResourceContents(uri, config), nil
    },
        mcpmux.WithResourceName("Configuration"),
        mcpmux.WithResourceDescription("Application configuration"),
        mcpmux.WithResourceMimeType("application/json"),
    )
    
    server.UseResourceProvider(resources)
}

func setupPrompts(server *mcpmux.Server) {
    prompts := mcpmux.NewStaticPromptProvider()
    
    // Code review prompt
    prompts.Register("code-review", func(ctx context.Context, args map[string]interface{}) (*mcpmux.GetPromptResult, error) {
        code, ok := args["code"].(string)
        if !ok {
            return nil, errors.New("code parameter required")
        }
        
        builder := mcpmux.NewPromptMessageBuilder()
        builder.AddSystemMessage(mcpmux.Content{
            Type: "text",
            Text: "You are a code reviewer. Analyze the provided code and suggest improvements.",
        })
        builder.AddUserMessage(mcpmux.Content{
            Type: "text",
            Text: "Please review this code:\n\n" + code,
        })
        
        return builder.Build("Code review prompt"), nil
    },
        mcpmux.WithPromptDescription("Reviews code and suggests improvements"),
        mcpmux.WithPromptArguments(
            mcpmux.PromptArg("code", "Code to review", true),
        ))
    
    server.UsePromptProvider(prompts)
}

func loggingMiddleware(next mcpmux.Handler) mcpmux.Handler {
    return func(ctx mcpmux.Context) error {
        method := ctx.Request().Method
        ctx.Logger().Info("Processing request", "method", method)
        return next(ctx)
    }
}

Core Concepts

Server

The Server is the main component that handles MCP protocol lifecycle, routing, and capabilities:

server := mcpmux.NewServer(
    mcpmux.WithServerInfo("my-server", "1.0.0"),
    mcpmux.WithInstructions("Server description"),
    mcpmux.WithLogging(),
)

Transports

mcpmux supports two transport mechanisms:

Stdio Transport

Perfect for CLI tools and subprocess communication:

// Use default stdio streams
transport := mcpmux.NewStdioTransport(nil)

// Or configure custom streams
transport := mcpmux.NewStdioTransport(&mcpmux.StdioTransportConfig{
    Stdin:  os.Stdin,
    Stdout: os.Stdout,
    Stderr: os.Stderr,
})

HTTP Transport with SSE

Ideal for web applications and network services:

// Use defaults (localhost:8080, path: /mcp)
transport := mcpmux.NewHTTPTransport(nil)

// Or configure custom settings
transport := mcpmux.NewHTTPTransport(&mcpmux.HTTPTransportConfig{
    Addr:           ":9090",
    Path:           "/api/mcp",
    AllowedOrigins: []string{"https://myapp.com"},
    SessionTimeout: 300, // seconds
    ResponseTimeout: 30, // seconds
})

Tools

Tools are functions that can be called by LLM applications:

tools := mcpmux.NewStaticToolAggregator()

// Simple tool
tools.Register("echo", func(ctx context.Context, args map[string]interface{}) (*mcpmux.ToolResult, error) {
    message := args["message"].(string)
    return mcpmux.TextResult(message), nil
}, mcpmux.WithDescription("Echoes the input message"))

// Tool with complex schema
tools.Register("analyze", analyzeFunction,
    mcpmux.WithDescription("Analyzes data"),
    mcpmux.WithParameters(
        mcpmux.StringParam("data", "Data to analyze", true),
        mcpmux.NumberParam("threshold", "Analysis threshold", false),
        mcpmux.BooleanParam("verbose", "Verbose output", false),
        mcpmux.ArrayParam("tags", "Analysis tags", false, map[string]interface{}{
            "type": "string",
        }),
    ))

server.UseToolAggregator(tools)

Tool Arguments (Args)

The Args type provides convenient methods for accessing tool arguments with type safety and default values:

tools.Register("profile", func(ctx context.Context, args mcpmux.Args) (*mcpmux.ToolResult, error) {
    // Get required arguments with validation
    if err := args.Required("name", "email").Validate(); err != nil {
        return mcpmux.ErrorResult(err), nil
    }
    
    // Get values with type conversion and defaults
    name := args.GetString("name")                    // Required, already validated
    email := args.GetString("email")                  // Required, already validated
    age := args.GetInt("age", 0)                     // Optional with default
    active := args.GetBool("active", true)           // Optional with default
    hobbies := args.GetStringSlice("hobbies", []string{}) // Optional array
    settings := args.GetMap("settings")              // Optional object
    
    // Check if optional arguments were provided
    if args.Has("phone") {
        phone := args.GetString("phone")
        // Handle phone number...
    }
    
    return mcpmux.TextResult(fmt.Sprintf("Profile: %s <%s>", name, email)), nil
})

// Alternative: Get required values directly with error handling
name, err := args.Required("name").GetString("name")
if err != nil {
    return mcpmux.ErrorResult(err), nil
}

Available getter methods:

• GetString(key, default...) - String values with optional default
• GetInt(key, default...) - Integer values with type conversion
• GetInt64(key, default...) - 64-bit integer values
• GetFloat64(key, default...) - Float values with type conversion
• GetBool(key, default...) - Boolean values with string parsing
• GetSlice(key, default...) - Array values as []interface{}
• GetStringSlice(key, default...) - String arrays with conversion
• GetMap(key, default...) - Object values as map[string]interface{}
• Has(key) - Check if key exists
• Get(key) - Get raw value as interface{}

Resources

Resources provide access to external data:

resources := mcpmux.NewStaticResourceProvider()

// File resource
resources.Register("file://data.json", func(ctx context.Context, uri string) ([]mcpmux.ResourceContents, error) {
    data, err := os.ReadFile("data.json")
    if err != nil {
        return nil, err
    }
    return mcpmux.JSONResourceContents(uri, string(data)), nil
})

// Dynamic resource with subscription support
resources.Register("live://metrics", func(ctx context.Context, uri string) ([]mcpmux.ResourceContents, error) {
    metrics := getCurrentMetrics()
    return mcpmux.TextResourceContents(uri, metrics), nil
},
    mcpmux.WithResourceName("Live Metrics"),
    mcpmux.WithResourceDescription("Real-time application metrics"),
)

server.UseResourceProvider(resources)

Prompts

Prompts are templates for LLM conversations:

prompts := mcpmux.NewStaticPromptProvider()

prompts.Register("summarize", func(ctx context.Context, args map[string]interface{}) (*mcpmux.GetPromptResult, error) {
    text := args["text"].(string)
    length := "short"
    if l, ok := args["length"].(string); ok {
        length = l
    }
    
    builder := mcpmux.NewPromptMessageBuilder()
    builder.AddSystemMessage(mcpmux.Content{
        Type: "text",
        Text: fmt.Sprintf("Summarize the following text in a %s format.", length),
    })
    builder.AddUserMessage(mcpmux.Content{
        Type: "text",
        Text: text,
    })
    
    return builder.Build("Text summarization prompt"), nil
},
    mcpmux.WithPromptDescription("Summarizes text content"),
    mcpmux.WithPromptArguments(
        mcpmux.PromptArg("text", "Text to summarize", true),
        mcpmux.PromptArg("length", "Summary length (short/medium/long)", false),
    ))

server.UsePromptProvider(prompts)

Middleware

Add any mcpmux middleware:

// MCP middleware (applies to all transports)
server.Use(func(next mcpmux.Handler) mcpmux.Handler {
    return func(ctx mcpmux.Context) error {
        start := time.Now()
        err := next(ctx)
        duration := time.Since(start)
        
        ctx.Logger().Info("Request processed",
            "method", ctx.Request().Method,
            "duration", duration,
            "error", err,
        )
        
        return err
    }
})

// Authentication middleware
server.Use(func(next mcpmux.Handler) mcpmux.Handler {
    return func(ctx mcpmux.Context) error {
        // Check authentication
        if !isAuthenticated(ctx) {
            return mcpmux.NewError(mcpmux.ErrorCodeInvalidRequest, "unauthorized", nil)
        }
        return next(ctx)
    }
})

HTTP Middleware

The HTTP transport supports standard Go HTTP middleware:

transport := mcpmux.NewHTTPTransport(&mcpmux.HTTPTransportConfig{
    Addr: ":8080",
    Middleware: []func(http.Handler) http.Handler{
        // Recovery middleware - catches panics
        mcpmux.RecoveryMiddleware(log.Default()),
        
        // Request ID middleware - adds X-Request-ID header
        mcpmux.RequestIDMiddleware("X-Request-ID"),
        
        // Logging middleware - logs HTTP requests
        mcpmux.LoggingMiddleware(log.Default()),
        
        // CORS middleware - handles cross-origin requests
        mcpmux.CORSMiddleware([]string{"http://localhost:3000"}),
        
        // Authentication middleware - validates requests  
        mcpmux.AuthMiddleware(func(r *http.Request) error {
            token := r.Header.Get("Authorization")
            if token == "" {
                return fmt.Errorf("missing authorization")
            }
            // Validate token...
            return nil
        }),
        
        // OR: Authentication with context modification
        mcpmux.AuthMiddlewareWithContext(func(r *http.Request) (*http.Request, error) {
            token := r.Header.Get("Authorization")
            if token == "" {
                return nil, fmt.Errorf("missing authorization")
            }
            
            // Add token to context for use in handlers
            ctx := context.WithValue(r.Context(), "auth-token", token)
            return r.WithContext(ctx), nil
        }),
        
        // Simple context middleware for static values
        mcpmux.ContextMiddleware("app-version", "1.0.0", "environment", "production"),
        
        // Timeout middleware - adds request timeout
        mcpmux.TimeoutMiddleware(30 * time.Second),
        
        // Custom middleware
        func(next http.Handler) http.Handler {
            return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
                // Custom logic before
                next.ServeHTTP(w, r)
                // Custom logic after
            })
        },
    },
})

The middlewares are executed in order of definition.

Custom Handlers

Register custom protocol handlers:

server.HandleFunc("custom/status", func(ctx mcpmux.Context) error {
    status := map[string]interface{}{
        "uptime": getUptime(),
        "memory": getMemoryUsage(),
        "connections": getConnectionCount(),
    }
    return ctx.Respond(status)
})

Advanced Usage

Custom Tool Aggregator

Implement your own tool aggregator for dynamic tools management:

type DatabaseToolAggregator struct {
    db *sql.DB
}

func (d *DatabaseToolAggregator) ListTools(ctx context.Context, cursor string) ([]mcpmux.Tool, string, error) {
    // Query tools from database
    rows, err := d.db.QueryContext(ctx, "SELECT name, description, schema FROM tools")
    if err != nil {
        return nil, "", err
    }
    defer rows.Close()
    
    var tools []mcpmux.Tool
    for rows.Next() {
        var tool mcpmux.Tool
        var schemaJSON string
        rows.Scan(&tool.Name, &tool.Description, &schemaJSON)
        
        json.Unmarshal([]byte(schemaJSON), &tool.InputSchema)
        tools = append(tools, tool)
    }
    
    return tools, "", nil
}

func (d *DatabaseToolAggregator) ExecuteTool(ctx context.Context, name string, args map[string]interface{}) (*mcpmux.ToolResult, error) {
    // Execute tool from database
    // Implementation details...
    return mcpmux.TextResult("Tool executed"), nil
}

// Use custom aggregator
server.UseToolAggregator(&DatabaseToolAggregator{db: db})

Progress Tracking

Support long-running operations with progress updates:

tools.Register("process-large-file", func(ctx context.Context, args map[string]interface{}) (*mcpmux.ToolResult, error) {
    filename := args["filename"].(string)
    file, err := os.Open(filename)
    if err != nil {
        return mcpmux.ErrorResult(err), nil
    }
    defer file.Close()
    
    // Get file size for progress calculation
    stat, _ := file.Stat()
    totalSize := stat.Size()
    
    processed := int64(0)
    buffer := make([]byte, 1024)
    
    for {
        n, err := file.Read(buffer)
        if err == io.EOF {
            break
        }
        if err != nil {
            return mcpmux.ErrorResult(err), nil
        }
        
        processed += int64(n)
        
        // Send progress update
        if mcpCtx, ok := ctx.(mcpmux.Context); ok {
            progress := float64(processed) / float64(totalSize) * 100
            mcpCtx.SendProgress(progress, 100)
        }
        
        // Process chunk...
        time.Sleep(10 * time.Millisecond) // Simulate processing
    }
    
    return mcpmux.TextResult("File processed successfully"), nil
})

Resource Subscriptions

Implement real-time resource updates:

resources.Register("live://logs", func(ctx context.Context, uri string) ([]mcpmux.ResourceContents, error) {
    logs := getCurrentLogs()
    return mcpmux.TextResourceContents(uri, logs), nil
})

// In your application, notify subscribers when logs change
go func() {
    for logUpdate := range logUpdateChan {
        resources.NotifyUpdate("live://logs", func(ctx context.Context) error {
            // Send update notification to client
            return nil
        })
    }
}()

Error Handling

mcpmux provides structured error handling following the JSON-RPC 2.0 specification:

// Standard MCP errors
return mcpmux.NewError(mcpmux.ErrorCodeResourceNotFound, "Resource not found", nil)
return mcpmux.NewError(mcpmux.ErrorCodeInvalidParams, "Missing required parameter", nil)

// Tool errors (returned as tool results, not protocol errors)
return mcpmux.ErrorResult(errors.New("Tool execution failed")), nil

// Custom errors with additional data
return mcpmux.NewError(-32001, "Custom error", map[string]interface{}{
    "code": "CUSTOM_ERROR",
    "details": "Additional error information",
}), nil

Testing

mcpmux includes comprehensive test utilities:

func TestMyServer(t *testing.T) {
    server := mcpmux.NewServer()
    
    // Register your handlers
    setupServer(server)
    
    // Create mock transport
    transport := newMockTransport()
    
    // Start server
    go server.Serve(transport)
    
    // Send test requests
    req := createRequest(1, "my/method", params)
    transport.receiveChan <- req
    
    // Verify responses
    resp := waitForResponse(t, transport, time.Second)
    // Assertions...
}

Examples

See the examples/ directory for complete working examples:

• Basic CLI Tool: Simple stdio-based MCP server
• Web Service: HTTP-based server with multiple capabilities
• Custom Providers: Implementing custom tool/resource/prompt providers
• Middleware: Using middleware for authentication and logging

Contributing

Contributions are welcome! Feel free to open a pull request and reach out to developers@kirha.com if you have any questions.

License

This project is licensed under the MIT License - see the LICENSE file for details.

Related Projects

• Model Context Protocol Specification
• MCP SDKs - Official MCP implementations

---

mcpmux - Build powerful MCP servers with ease! 🚀