pydantic-ai-go

pydantic-ai-go

Build typed LLM agents in Go.

pydantic-ai-go provides the agent loop, tools, structured output, streaming, usage limits, model middleware, and provider integrations. It follows PydanticAI's behavior while keeping the public API idiomatic for Go.

Install

You need Go 1.25 or newer.

go get github.com/Kludex/pydantic-ai-go/ai

Set the API key for your provider:

export OPENAI_API_KEY="..."

Create an agent with a tool

Create an Agent, register a typed function, then call Run:

package main

import (
	"context"
	"fmt"
	"log"

	ai "github.com/Kludex/pydantic-ai-go/ai"
	"github.com/Kludex/pydantic-ai-go/ai/models/openai"
)

type WeatherArgs struct {
	City string `json:"city" jsonschema:"description=City name"`
}

func main() {
	model := openai.NewModel("gpt-5-mini")
	agent := ai.NewAgent[struct{}, string](
		model,
		ai.WithInstructions("You are a concise weather assistant."),
	)

	ai.AddSimpleTool(agent, "get_weather", func(
		_ context.Context,
		args WeatherArgs,
	) (string, error) {
		return fmt.Sprintf("It is 21 C and sunny in %s.", args.City), nil
	}, ai.WithDescription("Get the current weather for a city"))

	result, err := agent.Run(
		context.Background(),
		"What is the weather in Berlin?",
		struct{}{},
	)
	if err != nil {
		log.Fatal(err)
	}

	fmt.Println(result.Output)
}

The type parameters are Agent[Deps, Output].

• Deps is application state available to tools and callbacks.
• Output is the final result type. Use string for text.
• WeatherArgs becomes the tool's JSON Schema.

The agent validates tool arguments before your function runs. Invalid arguments go back to the model as a retry prompt. Independent tool calls run concurrently and their results keep model order.

Use AddTool instead of AddSimpleTool when your function needs RunContext. The context contains typed dependencies, usage, history, retry state, and cancellation for the current run. Pass dependencies as the third argument to Run. One agent can serve concurrent runs safely.

Return structured data

Use a struct as the output type:

package main

import (
	"context"
	"fmt"
	"log"

	ai "github.com/Kludex/pydantic-ai-go/ai"
	"github.com/Kludex/pydantic-ai-go/ai/models/openai"
)

type City struct {
	Name       string `json:"name"`
	Country    string `json:"country"`
	Population int    `json:"population"`
}

func main() {
	agent := ai.NewAgent[struct{}, City](openai.NewModel("gpt-5-mini"))

	result, err := agent.Run(
		context.Background(),
		"Return information about Paris.",
		struct{}{},
	)
	if err != nil {
		log.Fatal(err)
	}

	fmt.Printf("%s, %s: %d\n", result.Output.Name, result.Output.Country, result.Output.Population)
}

The agent reflects a Draft 2020-12 JSON Schema from City. It validates the model response against that schema, decodes it into City, and retries invalid output.

OutputModeAuto is the default. It reads the selected model's ModelProfile; the standard profile chooses tool output because it works across providers. Use WithOutputMode(ai.OutputModeNative) to require native JSON Schema output, or OutputModePrompted to require JSON in text. See Structured output for validation and multiple output alternatives.

Stream a response

RunStream returns normalized events across all providers:

package main

import (
	"context"
	"fmt"
	"log"

	ai "github.com/Kludex/pydantic-ai-go/ai"
	"github.com/Kludex/pydantic-ai-go/ai/models/openai"
)

func main() {
	agent := ai.NewAgent[struct{}, string](openai.NewModel("gpt-5-mini"))
	stream := agent.RunStream(context.Background(), "Write a short poem.", struct{}{})

	for event, err := range stream.Events() {
		if err != nil {
			log.Fatal(err)
		}
		switch event := event.(type) {
		case ai.PartStartEvent:
			if text, ok := event.Part.(ai.TextPart); ok {
				fmt.Print(text.Content)
			}
		case ai.PartDeltaEvent:
			if text, ok := event.Delta.(ai.TextPartDelta); ok {
				fmt.Print(text.ContentDelta)
			}
		}
	}

	result := stream.Result()
	if result == nil {
		log.Fatal("stream ended without a result")
	}
	fmt.Printf("\nTokens: %d\n", result.Usage().TotalTokens())
}

Consume one stream view only. Use Events for lifecycle events or Outputs for typed partial output snapshots. Cancel the context to stop the request.

RunStream ends the run at the first output that matches the output type, so your consumer gets the final answer as soon as it starts. That output can be text, an output tool call, or a deferred tool call. When the same response also calls function tools, they still run, but their results are not sent back to the model. To run to completion while observing every event, register a listener with AddEventListener and call Run, or drive the run step by step with StartRun.

Continue a conversation

Pass the previous result's messages into the next run:

package main

import (
	"context"
	"fmt"
	"log"

	ai "github.com/Kludex/pydantic-ai-go/ai"
	"github.com/Kludex/pydantic-ai-go/ai/models/openai"
)

func main() {
	ctx := context.Background()
	agent := ai.NewAgent[struct{}, string](openai.NewModel("gpt-5-mini"))

	first, err := agent.Run(ctx, "My name is Ada.", struct{}{})
	if err != nil {
		log.Fatal(err)
	}
	second, err := agent.Run(
		ctx,
		"What is my name?",
		struct{}{},
		ai.WithMessageHistory(first.Messages()),
	)
	if err != nil {
		log.Fatal(err)
	}

	fmt.Println(second.Output)
}

Use MarshalMessages and UnmarshalMessages to persist history. Their JSON format is compatible with PydanticAI.

Choose a provider

The bundled providers use the same ai.Model interface.

Provider options configure caller-owned clients, endpoints, credentials, and default model settings. See Provider configuration for OpenAI-compatible endpoints and Azure API versions. See OpenAI Codex subscription for read-only CLI credentials, OAuth PKCE, and rotated credential storage. See Amazon Bedrock for AWS SDK configuration and Converse behavior.

Test an agent without network calls

Use ai/models/fakes in tests:

package weather_test

import (
	"context"
	"testing"

	ai "github.com/Kludex/pydantic-ai-go/ai"
	"github.com/Kludex/pydantic-ai-go/ai/models/fakes"
)

func TestAgent(t *testing.T) {
	agent := ai.NewAgent[struct{}, string](fakes.NewFunctionModel(func(
		_ context.Context,
		_ []ai.ModelMessage,
		_ ai.ModelRequestParams,
	) (*ai.ModelResponse, error) {
		return &ai.ModelResponse{
			Parts: []ai.ResponsePart{ai.TextPart{Content: "hello"}},
		}, nil
	}))

	result, err := agent.Run(t.Context(), "Say hello.", struct{}{})
	if err != nil {
		t.Fatal(err)
	}
	if result.Output != "hello" {
		t.Fatalf("got %q", result.Output)
	}
}

fakes.NewFunctionModel gives you complete control over each model response. fakes.NewTestModel can generate schema-valid calls for every registered tool.

API guide

Run options are detached from agent configuration. They are safe to use in concurrent runs. See the Go package documentation for each public contract and CHECKLIST.md for remaining parity work.

Design

The agent is a plain loop:

1. Build a model request.
2. Ask the model for a response.
3. Validate and execute tool calls.
4. Repeat until a final output is valid.

Capabilities intercept these semantic boundaries. The package does not expose an internal graph API. This keeps the control flow readable and lets Go interfaces provide small extension points.

Per-run settings, tools, capabilities, output types, histories, and provider details are copied before use. One run cannot mutate another run's configuration.

Development

Run the complete local checks:

gofmt -w $(git ls-files '*.go')
go vet ./...
golangci-lint run ./...
go test ./...
scripts/coverage.sh
go test -race ./...
go test -run '^$' -bench . -benchmem ./ai

The project requires 100% statement coverage for every tested package. See the benchmark guide to compare performance changes.

Status

The API is under active development while it approaches feature parity with PydanticAI. Review CHECKLIST.md before depending on a provider-specific or advanced feature.

See the COMPATIBILITY.md version guarantees and CHANGELOG.md before you upgrade.

License

This project is licensed under the MIT License. See LICENSE.