Axonflow Sdk Go

AxonFlow SDK for Go

[](https://pkg.go.dev/github.com/getaxonflow/axonflow-sdk-go/v8) [](https://goreportcard.com/report/github.com/getaxonflow/axonflow-sdk-go) [](https://opensource.org/licenses/MIT)

> Upgrade strongly recommended. AxonFlow ships substantial monthly security and quality hardening; staying on the latest major is the security-supported release line. Latest release · Security advisories

> ## ⚠️ Use the /v8 import path > > Go's semantic import versioning requires the module path to include the major version suffix for v2+. The current release line is v8.x, imported as: > > ``go > import "github.com/getaxonflow/axonflow-sdk-go/v8" > ` > > `bash > go get github.com/getaxonflow/axonflow-sdk-go/v8 > ` > > go get github.com/getaxonflow/axonflow-sdk-go@latest (without /v8`) resolves to v1.17.0 (a 2026-01 relic from before the v2 split) and is seven major release lines behind current. See the [Migration Guide](#migration-guide) below.

> Evaluating AxonFlow for a real deployment? > > Choose the path that fits: > - Self-serve: free 90-day Evaluation License > - Hands-on: Design Partner Program — Enterprise access for the scoped engagement, founder-led architecture and rollout support, and preferential pricing after successful rollout > > Priority support, architecture review, incident-readiness review, and roadmap input are included for selected partners. We reply within 48 hours.

> Questions or feedback? > > Comment in GitHub Discussions or email [hello@getaxonflow.com](mailto:hello@getaxonflow.com) for private feedback.

Enterprise-grade Go SDK for AxonFlow AI governance platform. Add invisible AI governance to your applications with production-ready features including retry logic, caching, fail-open strategy, and debug mode.

How This SDK Fits with AxonFlow

This SDK is a client library for interacting with a running AxonFlow control plane. It is used from application or agent code to send execution context, policies, and requests at runtime.

A deployed AxonFlow platform (self-hosted or cloud) is required for end-to-end AI governance. SDKs alone are not sufficient—the platform and SDKs are designed to be used together.

See AxonFlow in Action

Three short videos covering different angles of the platform:

• Community Quickstart Demo (Code + Terminal, 2.5 min) — governed calls, PII block, Gateway Mode with LangChain/CrewAI, and MAP from YAML
• Runtime Control Demo (Portal + Workflow, 3 min) — approvals, retry safety, execution state, and the audit viewer
• Architecture Deep Dive (12 min) — how the control plane works, policy enforcement flow, and multi-agent planning

Installation

go get github.com/getaxonflow/axonflow-sdk-go/v8

Evaluation Tier (Free License)

Need more capacity than Community without moving to Enterprise? Evaluation uses the same core features with higher limits:

| Limit | Community | Evaluation (Free) | Enterprise | |-------|-----------|-------------------|------------| | Tenant policies | 20 | 50 | Unlimited | | Org-wide policies | 0 | 5 | Unlimited | | Audit retention | 3 days | 14 days | 3650 days | | Concurrent executions | 5 | 25 | Unlimited | | Pending execution approvals | 5 | 25 | Unlimited | | Evidence export (CSV / JSON) | — | 5,000 records · 14d window · 3/day | Unlimited | | Policy simulation | — | 300 / day | Unlimited |

Concurrent executions applies to MAP and WCP executions per tenant. Pending execution approvals applies to MAP confirm/step mode and WCP approval queues.

> Note: Evidence export and policy simulation are licensed AxonFlow platform capabilities available alongside the SDK on your deployed platform — not language-specific SDK helpers. Access them via the platform API or customer portal. The SDK row is included to show what your licensed deployment unlocks at each tier.

Get a free Evaluation license · Apply for Design Partner · Full feature matrix

Try Without Installing

Skip local setup entirely — try AxonFlow instantly at try.getaxonflow.com:

# 1. Register (30 seconds)
curl -X POST https://try.getaxonflow.com/api/v1/register \
  -H "Content-Type: application/json" -d '{"label":"my-trial"}'

# 2. Set credentials and auto-connect
export AXONFLOW_TRY=1
export AXONFLOW_CLIENT_ID=cs_your-tenant-id
export AXONFLOW_CLIENT_SECRET=your-secret

No Docker, no license, no installation. Rate-limited to 20 req/min. Learn more.

Quick Start

Basic Usage (OAuth2 Client Credentials)

package main

import (
    "fmt"
    "log"
    "os"

    "github.com/getaxonflow/axonflow-sdk-go/v8"
)

func main() {
    // Simple initialization with OAuth2 credentials
    client := axonflow.NewClient(axonflow.AxonFlowConfig{
        Endpoint:     "http://localhost:8080",
        ClientID:     os.Getenv("AXONFLOW_CLIENT_ID"),
        ClientSecret: os.Getenv("AXONFLOW_CLIENT_SECRET"),
    })

    // Execute a governed query
    resp, err := client.ProxyLLMCall(
        "user-token",
        "What is the capital of France?",
        "chat",
        map[string]interface{}{},
    )

    if err != nil {
        log.Fatalf("Query failed: %v", err)
    }

    if resp.Blocked {
        log.Printf("Request blocked: %s", resp.BlockReason)
        return
    }

    fmt.Printf("Result: %s\n", resp.Data)
}

Advanced Configuration

import (
    "time"
    "os"
    "github.com/getaxonflow/axonflow-sdk-go/v8"
)

// Full configuration with all features
client := axonflow.NewClient(axonflow.AxonFlowConfig{
    Endpoint:     "http://localhost:8080",
    ClientID:     os.Getenv("AXONFLOW_CLIENT_ID"),
    ClientSecret: os.Getenv("AXONFLOW_CLIENT_SECRET"),
    Mode:         "production",  // or "sandbox"
    Debug:        true,          // Enable debug logging
    Timeout:      60 * time.Second,

    // Retry configuration (exponential backoff)
    Retry: axonflow.RetryConfig{
        Enabled:      true,
        MaxAttempts:  3,
        InitialDelay: 1 * time.Second,
    },

    // Cache configuration (in-memory with TTL)
    Cache: axonflow.CacheConfig{
        Enabled: true,
        TTL:     60 * time.Second,
    },
})

Self-Hosted Mode (No License Required)

Connect to a self-hosted AxonFlow instance running via docker-compose:

package main

import (
    "fmt"
    "log"

    "github.com/getaxonflow/axonflow-sdk-go/v8"
)

func main() {
    // Self-hosted (localhost) - no license key needed!
    client := axonflow.NewClient(axonflow.AxonFlowConfig{
        Endpoint: "http://localhost:8081",
        // That's it - no authentication required for localhost
    })

    // Use normally - same features as production
    resp, err := client.ProxyLLMCall(
        "user-token",
        "Test with self-hosted AxonFlow",
        "chat",
        map[string]interface{}{},
    )

    if err != nil {
        log.Fatalf("Query failed: %v", err)
    }

    fmt.Printf("Result: %s\n", resp.Data)
}

Self-hosted deployment:

# Clone and start AxonFlow
git clone https://github.com/getaxonflow/axonflow.git
cd axonflow
export OPENAI_API_KEY=sk-your-key-here
docker-compose up

# Go SDK connects to http://localhost:8081 - no license needed!

Features:

• ✅ Full AxonFlow features without license
• ✅ Perfect for local development and testing
• ✅ Same API as production
• ✅ Automatically detects localhost and skips authentication

Sandbox Mode (Local Testing)

// Quick sandbox client for local testing — defaults to http://localhost:8080.
client := axonflow.Sandbox("demo-client", "demo-secret")

resp, err := client.ProxyLLMCall(
    "",
    "Test query with sensitive data: SSN 123-45-6789",
    "chat",
    map[string]interface{}{},
)

if resp.Blocked {
    fmt.Printf("Blocked: %s\n", resp.BlockReason)
}

> Sandbox-mode clients fire telemetry like every other client — anonymous SDK > heartbeat, classification-only payload, opt-out via AXONFLOW_TELEMETRY=off. > Pings are tagged stream="sandbox" server-side so dev/test usage is > distinguishable from production heartbeat.

Features

✅ Retry Logic with Exponential Backoff

Automatic retry on transient failures with exponential backoff:

client := axonflow.NewClient(axonflow.AxonFlowConfig{
    Endpoint: "http://localhost:8080",
    ClientID: "your-client-id",
    ClientSecret: "your-secret",
    Retry: axonflow.RetryConfig{
        Enabled:      true,
        MaxAttempts:  3,           // Retry up to 3 times
        InitialDelay: 1 * time.Second,  // 1s, 2s, 4s backoff
    },
})

// Automatically retries on 5xx errors or network failures
resp, err := client.ProxyLLMCall(...)

✅ In-Memory Caching with TTL

Reduce latency and load with intelligent caching:

client := axonflow.NewClient(axonflow.AxonFlowConfig{
    Endpoint: "http://localhost:8080",
    ClientID: "your-client-id",
    ClientSecret: "your-secret",
    Cache: axonflow.CacheConfig{
        Enabled: true,
        TTL:     60 * time.Second,  // Cache for 60 seconds
    },
})

// First call: hits AxonFlow
resp1, _ := client.ProxyLLMCall("token", "query", "chat", nil)

// Second call (within 60s): served from cache
resp2, _ := client.ProxyLLMCall("token", "query", "chat", nil)

✅ Fail-Open Strategy (Production Mode)

Never block your users if AxonFlow is unavailable:

client := axonflow.NewClient(axonflow.AxonFlowConfig{
    Endpoint: "http://localhost:8080",
    ClientID: "your-client-id",
    ClientSecret: "your-secret",
    Mode:     "production",  // Fail-open in production
    Debug:    true,
})

// If AxonFlow is unavailable, request proceeds with warning
resp, err := client.ProxyLLMCall(...)
// err == nil, resp.Success == true, resp.Error contains warning

LLM Interceptors (OpenAI & Anthropic)

Wrap your LLM clients with automatic AxonFlow governance using the interceptors package:

OpenAI Interceptor

import (
    "context"
    "github.com/sashabaranov/go-openai"
    "github.com/getaxonflow/axonflow-sdk-go/v8"
    "github.com/getaxonflow/axonflow-sdk-go/v8/interceptors"
)

// Initialize AxonFlow client
axonflowClient := axonflow.NewClient(axonflow.AxonFlowConfig{
    Endpoint:     "http://localhost:8080",
    ClientID:     os.Getenv("AXONFLOW_CLIENT_ID"),
    ClientSecret: os.Getenv("AXONFLOW_CLIENT_SECRET"),
})

// Create an adapter for the OpenAI client
openaiClient := openai.NewClient(os.Getenv("OPENAI_API_KEY"))

// Use the function wrapper for direct usage
wrappedFn := interceptors.WrapOpenAIFunc(
    func(ctx context.Context, req interceptors.ChatCompletionRequest) (interceptors.ChatCompletionResponse, error) {
        // Convert to go-openai types and call
        goReq := openai.ChatCompletionRequest{
            Model: req.Model,
            Messages: convertMessages(req.Messages),
        }
        resp, err := openaiClient.CreateChatCompletion(ctx, goReq)
        if err != nil {
            return interceptors.ChatCompletionResponse{}, err
        }
        return convertResponse(resp), nil
    },
    axonflowClient,
    "user-token",
)

// Use wrapped function - governance happens automatically
resp, err := wrappedFn(ctx, interceptors.ChatCompletionRequest{
    Model: "gpt-4",
    Messages: []interceptors.ChatMessage{
        {Role: "user", Content: "Hello, world!"},
    },
})

if err != nil {
    if interceptors.IsPolicyViolationError(err) {
        pve, _ := interceptors.GetPolicyViolation(err)
        log.Printf("Blocked: %s (policies: %v)", pve.BlockReason, pve.Policies)
    }
}

Anthropic Interceptor

import (
    "context"
    "github.com/getaxonflow/axonflow-sdk-go/v8"
    "github.com/getaxonflow/axonflow-sdk-go/v8/interceptors"
)

// Create Anthropic interceptor
wrappedFn := interceptors.WrapAnthropicFunc(
    yourAnthropicCreateFn,
    axonflowClient,
    "user-token",
)

// Use wrapped function
resp, err := wrappedFn(ctx, interceptors.AnthropicMessageRequest{
    Model:     "claude-3-sonnet-20240229",
    MaxTokens: 1024,
    Messages: []interceptors.AnthropicMessage{
        interceptors.CreateUserMessage("Hello, Claude!"),
    },
})

Interface-Based Wrapping

For more flexibility, implement the OpenAIChatCompleter or AnthropicMessageCreator interfaces:

// Implement the interface
type MyOpenAIClient struct {
    // your fields
}

func (c *MyOpenAIClient) CreateChatCompletion(ctx context.Context, req interceptors.ChatCompletionRequest) (interceptors.ChatCompletionResponse, error) {
    // your implementation
}

// Wrap the client
wrapped := interceptors.WrapOpenAIClient(&MyOpenAIClient{}, axonflowClient, "user-token")

// Use wrapped client
resp, err := wrapped.CreateChatCompletion(ctx, req)

MCP Connector Marketplace

Integrate with external data sources using AxonFlow's MCP (Model Context Protocol) connectors:

List Available Connectors

connectors, err := client.ListConnectors()
if err != nil {
    log.Fatalf("Failed to list connectors: %v", err)
}

for _, conn := range connectors {
    fmt.Printf("Connector: %s (%s)\n", conn.Name, conn.Type)
    fmt.Printf("  Description: %s\n", conn.Description)
    fmt.Printf("  Installed: %v\n", conn.Installed)
}

Install a Connector

err := client.InstallConnector(axonflow.ConnectorInstallRequest{
    ConnectorID: "amadeus-travel",
    Name:        "amadeus-prod",
    TenantID:    "your-tenant-id",
    Options: map[string]interface{}{
        "environment": "production",
    },
    Credentials: map[string]string{
        "api_key":    "your-amadeus-key",
        "api_secret": "your-amadeus-secret",
    },
})

if err != nil {
    log.Fatalf("Failed to install connector: %v", err)
}

fmt.Println("Connector installed successfully!")

Query a Connector

// Query the Amadeus connector for flight information
resp, err := client.QueryConnector(
    "user-session-token",  // User token for authentication and audit trail
    "amadeus-prod",
    "Find flights from Paris to Amsterdam on Dec 15",
    map[string]interface{}{
        "origin":      "CDG",
        "destination": "AMS",
        "date":        "2025-12-15",
    },
)

if err != nil {
    log.Fatalf("Connector query failed: %v", err)
}

if resp.Success {
    fmt.Printf("Flight data: %v\n", resp.Data)
} else {
    fmt.Printf("Query failed: %s\n", resp.Error)
}

Production Connectors (November 2025)

AxonFlow now supports 7 production-ready connectors:

Salesforce CRM Connector

Query Salesforce data using SOQL:

// Query Salesforce contacts
resp, err := client.QueryConnector(
    "user-session-token",  // User token for authentication and audit trail
    "salesforce-crm",
    "Find all contacts for account Acme Corp",
    map[string]interface{}{
        "soql": "SELECT Id, Name, Email, Phone FROM Contact WHERE AccountId = '001xx000003DHP0'",
    },
)

if err != nil {
    log.Fatalf("Salesforce query failed: %v", err)
}

fmt.Printf("Found %d contacts\n", len(resp.Data.([]interface{})))

Authentication: OAuth 2.0 password grant (configured in AxonFlow dashboard)

Snowflake Data Warehouse Connector

Execute analytics queries on Snowflake:

// Qu

…

## Source & license

This open-source MCP server is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [getaxonflow](https://github.com/getaxonflow)
- **Source:** [getaxonflow/axonflow-sdk-go](https://github.com/getaxonflow/axonflow-sdk-go)
- **License:** MIT

Install and usage instructions live in the source repository linked above.