mcptoolkit

mcp-toolkit

A Go library for building OpenAI-compatible function-calling tools with typed handlers, retry-aware execution, and first-class Model Context Protocol (MCP) support — both as a client and as a server.

The struct is the single source of truth for tool schemas. Adding or removing a field automatically updates the JSON Schema sent to the LLM with no manual synchronisation.

Install

go get github.com/v8tix/mcp-toolkit

Requires Go 1.26+.

At a Glance

// 1. Define args
type SearchArgs struct {
    Query string `json:"query" desc:"The search query."`
    Limit *int   `json:"limit,omitempty" desc:"Maximum results."`
}

// 2. Create a typed tool
tool := handler.NewTool("search_web", "Search the web.", func(ctx context.Context, in SearchArgs) ([]Result, error) {
    return repo.Search(ctx, in.Query)
})

// 3. Register and send to the LLM
reg := registry.New(tool)
req.Tools = reg.All()

// 4. Dispatch after the LLM responds
t, _ := reg.ByName(call.Function.Name)
result, err := t.(handler.ExecutableTool).Execute(ctx, call.Function.Arguments)

Table of Contents

• Packages
• model
• schema
• handler
• registry
• observable
• mcp
  • Client — consume an MCP server's tools
  • Server — expose a registry as an MCP server
• Full example — typed tools + registry + MCP server
• Full example — consume MCP tools + retry + registry
• Test coverage
• License

Packages

model

Core data types shared by all packages. Import directly only when defining custom Tool implementations.

import "github.com/v8tix/mcp-toolkit/model"

InputSchema.ToMap() converts the schema to map[string]any — used when registering tools on an MCP server.

schema

Derives InputSchema and ToolDefinition from Go struct tags at construction time.

import "github.com/v8tix/mcp-toolkit/schema"

Struct tags

Required vs optional

Go → JSON Schema type mapping

Builder functions

// Derive schema from struct tags — panics if v is not a struct.
schema.NewInputSchemaFromStruct(v any) model.InputSchema

// Build a full ToolDefinition with explicit strict control.
schema.FormatToolDefinition(name, description string, params model.InputSchema, strict bool) model.ToolDefinition

// Convenience: derives schema and always enables strict mode.
schema.NewStrictTool(name, description string, args any) model.ToolDefinition

strict: false disables OpenAI structured-outputs enforcement. Use it when your schema has fields that cannot satisfy strict mode requirements (e.g. anyOf, oneOf, or additional properties).

Example — schema only (no handler)

type SearchArgs struct {
    Query string `json:"query" desc:"Search query."`
    Topic string `json:"topic,omitempty" desc:"Category." enum:"general,news,finance"`
}

// Derive and send directly to the LLM request.
def := schema.NewStrictTool("search_web", "Search the web.", SearchArgs{})
req.Tools = []model.ToolDefinition{def}

handler

Creates typed, executable tools from Go functions. The JSON Schema is derived once at construction time from the In struct.

import "github.com/v8tix/mcp-toolkit/handler"

Creating tools

// Derive schema from In struct tags (always strict).
tool := handler.NewTool("search_web", "Search the web.",
    func(ctx context.Context, in SearchArgs) ([]Result, error) {
        return repo.Search(ctx, in.Query)
    },
)

// Use a pre-built definition — for non-strict mode, custom schemas, or
// definitions that cannot be expressed via struct tags alone.
def := schema.FormatToolDefinition("search_web", "Search.", params, false)
tool := handler.NewToolWithDefinition(def,
    func(ctx context.Context, in SearchArgs) ([]Result, error) {
        return repo.Search(ctx, in.Query)
    },
)

Interfaces

Middleware

Wrap a tool to add cross-cutting behaviour without modifying the handler:

type ToolMiddleware func(ctx context.Context, rawArgs json.RawMessage, next ExecuteFunc) (any, error)

tool = handler.Wrap(tool, func(ctx context.Context, rawArgs json.RawMessage, next handler.ExecuteFunc) (any, error) {
    log.Printf("call: %s", rawArgs)
    result, err := next(ctx, rawArgs)
    if err != nil {
        log.Printf("error: %v", err)
    }
    return result, err
})

Decorators stack outermost-first:

tool = handler.Wrap(tool, withTimeout(5*time.Second)) // runs second
tool = handler.Wrap(tool, withLogging(log.Printf))    // runs first

Errors

handler.ErrInvalidArguments is returned by Execute when the raw JSON arguments cannot be unmarshalled into the handler's input type. The observable layer treats this as a permanent error and never retries it.

result, err := t.(handler.ExecutableTool).Execute(ctx, call.Function.Arguments)
if errors.Is(err, handler.ErrInvalidArguments) {
    // bad args from the LLM — log and skip, do not retry
}

registry

Thread-safe, ordered catalogue of model.Tool objects. Decoupled from execution — callers type-assert to handler.ExecutableTool at dispatch time.

import "github.com/v8tix/mcp-toolkit/registry"

Creating and populating

reg := registry.New(searchTool, translateTool)
reg.Add(anotherTool).Add(yetAnotherTool) // fluent chaining

Reading

// All definitions in insertion order — pass to LLM request.
req.Tools = reg.All()

// Look up by name, then execute.
t, ok := reg.ByName(call.Function.Name)
result, err := t.(handler.ExecutableTool).Execute(ctx, call.Function.Arguments)

// Ordered name list.
fmt.Println(reg.Names()) // ["search_web", "translate", ...]

Mutating

// Remove a tool by name. Returns true if it existed.
removed := reg.Remove("search_web")

// Create a sub-registry by predicate — original is unchanged.
execOnly := reg.Filter(func(t model.Tool) bool {
    _, ok := t.(handler.ExecutableTool)
    return ok
})

publicTools := reg.Filter(func(t model.Tool) bool {
    return !strings.HasPrefix(t.Definition().Function.Name, "internal_")
})

Key properties

observable

Wraps any ExecutableTool in a retry-aware reactive layer using rxgo and cenkalti/backoff.

import "github.com/v8tix/mcp-toolkit/observable"

Retry lives in ExecuteRx. The synchronous Execute path skips retry — use ExecuteRx when you need backoff behaviour.

Creating observable tools

// Primary entry point — typed handler, options applied at construction.
// Go infers [In, Out] from the function signature.
tool := observable.New("search_web", "Search.", myFn)
tool := observable.New("search_web", "Search.", myFn, observable.WithMaxRetries(5))

// Wrap an existing ExecutableTool.
tool := observable.Wrap(handlerTool, observable.WithMaxRetries(3))

Options

tool := observable.New("search_web", "Search.", myFn,
    observable.WithMaxRetries(5),
    observable.WithClassifier(func(err error) error {
        if errors.Is(err, ErrNotFound) {
            return observable.Permanent(err) // stop retrying immediately
        }
        return err
    }),
    observable.WithOnRetry(func(attempt uint64, err error) {
        log.Printf("retry %d after: %v", attempt, err)
    }),
)

Permanent errors

Use observable.Permanent(err) anywhere to mark an error as non-retryable:

// In a handler:
if in.Query == "" {
    return nil, observable.Permanent(ErrEmptyQuery)
}

// In a classifier:
observable.WithClassifier(func(err error) error {
    if errors.Is(err, ErrNotFound) { return observable.Permanent(err) }
    return err
})

errors.Is / errors.As still work through the wrapper — backoff.PermanentError implements Unwrap.

Custom retry policy

Implement RetryPolicy to plug in a circuit breaker or fixed interval:

type RetryPolicy interface {
    MaxRetries() uint64
    NewBackOff() backoff.BackOff
}

type fixedRetryPolicy struct{}
func (fixedRetryPolicy) MaxRetries() uint64         { return 3 }
func (fixedRetryPolicy) NewBackOff() backoff.BackOff { return backoff.NewConstantBackOff(500 * time.Millisecond) }

tool := observable.New("t", "d", myFn, observable.WithRetryPolicy(fixedRetryPolicy{}))

Custom error policy

Implement ErrorPolicy for full control over classification logic:

type ErrorPolicy interface {
    Classify(err error) error
}

WithClassifier is a convenience wrapper for the common case where a single function is enough. WithErrorPolicy accepts any ErrorPolicy implementation for stateful classifiers (e.g. one that tracks error counts to trip a circuit breaker).

Executing the observable

var result any
var execErr error
for item := range tool.ExecuteRx(ctx, rawArgs).Observe() {
    if item.E != nil {
        execErr = item.E
    } else {
        result = item.V
    }
}

mcp

Bridges the Model Context Protocol Go SDK with mcp-toolkit in both directions.

import "github.com/v8tix/mcp-toolkit/mcp"

Client — consume an MCP server's tools

Wrap tools discovered from an MCP server as retry-aware observable.Tool objects and drop them into a registry:

session, _ := mcpClient.Connect(ctx, transport, nil)
toolsResult, _ := session.ListTools(ctx, nil)

// Zero options — production defaults (3 retries, exponential backoff).
reg := registry.New(mcp.NewTools(toolsResult.Tools, session).Build()...)

NewTool — single tool builder

tool := mcp.NewTool(discovered, session)

NewTools — batch builder

tools := mcp.NewTools(discovered, session).WithMaxRetries(0).Build()
reg := registry.New(tools...)

Builder options

All builder methods return a new builder (immutable chaining). sync.Once lazy-resolves the final observable.Tool on first use.

// Override description for all tools (e.g. add a prefix for caching).
tools := mcp.NewTools(discovered, session).
    WithDefinitionFunc(func(t *sdkmcp.Tool) model.ToolDefinition {
        def := mcp.BuildDefinition(t) // use your own helper or the default
        def.Function.Description = "[cached] " + def.Function.Description
        return def
    }).
    Build()

// Single tool — custom retry + classifier.
tool := mcp.NewTool(discovered[0], session).
    WithMaxRetries(5).
    WithClassifier(func(err error) error {
        if errors.Is(err, ErrRateLimit) { return observable.Permanent(err) }
        return err
    })

// WithOnRetry is an observable.Option — pass it via With.
tools := mcp.NewTools(discovered, session).
    WithMaxRetries(3).
    With(observable.WithOnRetry(func(attempt uint64, err error) {
        log.Printf("retry %d: %v", attempt, err)
    })).
    Build()

Session interface

Any type with a CallTool method satisfies mcp.Session. *sdkmcp.ClientSession satisfies it out of the box. Implement the interface to add caching, logging, or request transformation:

type Session interface {
    CallTool(ctx context.Context, params *sdkmcp.CallToolParams) (*sdkmcp.CallToolResult, error)
}

// Example: logging wrapper
type loggingSession struct {
    inner mcp.Session
    log   *log.Logger
}

func (s *loggingSession) CallTool(ctx context.Context, params *sdkmcp.CallToolParams) (*sdkmcp.CallToolResult, error) {
    s.log.Printf("calling tool %q", params.Name)
    res, err := s.inner.CallTool(ctx, params)
    if err != nil {
        s.log.Printf("tool %q error: %v", params.Name, err)
    }
    return res, err
}

Server — expose a registry as an MCP server

Register every tool in a registry.Registry on an *sdkmcp.Server with a single call:

s := sdkmcp.NewServer(&sdkmcp.Implementation{Name: "my-agent", Version: "v1.0.0"}, nil)
mcp.RegisterTools(s, reg)

http.Handle("/mcp", sdkmcp.NewStreamableHTTPHandler(
    func(_ *http.Request) *sdkmcp.Server { return s },
    &sdkmcp.StreamableHTTPOptions{Stateless: true},
))

Selective exposure — filter tools

Pass one or more filter functions to expose only a subset of the registry:

// Expose only tools whose names don't start with "internal_".
mcp.RegisterTools(s, reg, func(t model.Tool) bool {
    return !strings.HasPrefix(t.Definition().Function.Name, "internal_")
})

// Multiple filters — tool must pass all of them.
mcp.RegisterTools(s, reg, isPublic, isStable)

Key properties

Full example — typed tools + registry + MCP server

package main

import (
    "context"
    "net/http"

    sdkmcp "github.com/modelcontextprotocol/go-sdk/mcp"
    "github.com/v8tix/mcp-toolkit/handler"
    "github.com/v8tix/mcp-toolkit/mcp"
    "github.com/v8tix/mcp-toolkit/registry"
)

type SearchArgs struct {
    Query string `json:"query" desc:"The search query."`
    Limit *int   `json:"limit,omitempty" desc:"Maximum results."`
}

type SearchResult struct {
    URL   string `json:"url"`
    Title string `json:"title"`
}

type TranslateArgs struct {
    Text       string `json:"text" desc:"Text to translate."`
    TargetLang string `json:"target_lang" desc:"Target language code." enum:"en,es,fr,de"`
}

func main() {
    reg := registry.New(
        handler.NewTool("search_web", "Search the web.", func(ctx context.Context, in SearchArgs) ([]SearchResult, error) {
            return searchService.Search(ctx, in.Query)
        }),
        handler.NewTool("translate", "Translate text.", func(ctx context.Context, in TranslateArgs) (string, error) {
            return translateService.Translate(ctx, in.Text, in.TargetLang)
        }),
    )

    s := sdkmcp.NewServer(&sdkmcp.Implementation{Name: "my-agent", Version: "v1.0.0"}, nil)
    mcp.RegisterTools(s, reg)

    http.Handle("/mcp", sdkmcp.NewStreamableHTTPHandler(
        func(_ *http.Request) *sdkmcp.Server { return s },
        &sdkmcp.StreamableHTTPOptions{Stateless: true},
    ))
    http.ListenAndServe(":8080", nil)
}

Full example — consume MCP tools + retry + registry

package main

import (
    "context"
    "log"

    sdkmcp "github.com/modelcontextprotocol/go-sdk/mcp"
    "github.com/v8tix/mcp-toolkit/mcp"
    "github.com/v8tix/mcp-toolkit/observable"
    "github.com/v8tix/mcp-toolkit/registry"
)

func main() {
    client := sdkmcp.NewClient(&sdkmcp.Implementation{Name: "my-client", Version: "v1"}, nil)

    // transport is your stdio/HTTP/SSE transport (see MCP Go SDK docs).
    session, _ := client.Connect(context.Background(), transport, nil)

    toolsResult, _ := session.ListTools(context.Background(), nil)

    reg := registry.New(
        mcp.NewTools(toolsResult.Tools, session).
            WithMaxRetries(3).
            With(observable.WithOnRetry(func(attempt uint64, err error) {
                log.Printf("retry %d: %v", attempt, err)
            })).
            Build()...,
    )

    // Use reg.All() in your LLM request, reg.ByName() for dispatch.
}

Test coverage

Run tests:

go test -race ./...

License

MIT — see LICENSE.