Ollama plugin

The genkit-ollama package connects Genkit to a local or remote Ollama server. Any model or embedder the server has pulled resolves as 'ollama/<name>', with its tool and vision support read from the server.

Prerequisites

Install and start Ollama from ollama.com/download, then pull the models you want to call:

ollama pull llama3.2
ollama pull qwen3
ollama pull nomic-embed-text

Installation

uv add genkit-ollama

Configuration

Ollama() connects to OLLAMA_HOST when it is set, otherwise to http://127.0.0.1:11434. There is no model list to register; name any pulled model as 'ollama/<name>':

from genkit import Genkit
from genkit_ollama import Ollama

ai = Genkit(
    plugins=[Ollama()],
    model='ollama/llama3.2',
)

Custom endpoints and headers

Pass base_url, request_headers, and timeout (in seconds) when pointing at a remote or authenticated Ollama deployment. base_url is the server root: requests go to {base_url}/api/chat, so a URL ending in /v1 or /api raises ValueError.

import os
from genkit import Genkit
from genkit_ollama import Ollama

ai = Genkit(
    plugins=[
        Ollama(
            base_url='https://ollama.internal:11434',
            request_headers={'Authorization': f"Bearer {os.environ['OLLAMA_TOKEN']}"},
            timeout=60.0,
        )
    ],
)

When request_headers sets no Authorization header, the Ollama client sends the OLLAMA_API_KEY environment variable as a bearer token, so hosted Ollama needs only Ollama(base_url='https://ollama.com'). To bill a different key per call, pass context={'secrets': {'api_key': tenant_key}} to ai.generate(); that key authenticates the call only and stays out of traces.

For a token that expires, pass a callable instead of a dict. It takes no arguments, can be sync or async, and runs before every request, so it should cache the token itself. This one sends a Google ID token to an Ollama server on Cloud Run:

import asyncio

import google.auth.transport.requests
import google.oauth2.id_token
from genkit import Genkit
from genkit_ollama import Ollama

OLLAMA_URL = 'https://ollama-gpu-abc123.a.run.app'
credentials = google.oauth2.id_token.fetch_id_token_credentials(OLLAMA_URL)

async def cloud_run_headers() -> dict[str, str]:
    # Refresh off the event loop, and only when the cached token has expired.
    if not credentials.valid:
        await asyncio.to_thread(credentials.refresh, google.auth.transport.requests.Request())
    return {'Authorization': f'Bearer {credentials.token}'}

ai = Genkit(plugins=[Ollama(base_url=OLLAMA_URL, request_headers=cloud_run_headers)])

An unreachable server doesn’t make ai.generate() raise. The response comes back with response.error.status set to UNAVAILABLE, or DEADLINE_EXCEEDED when timeout runs out. See Error types.

Usage

from genkit import Genkit
from genkit_ollama import Ollama, OllamaConfig

ai = Genkit(plugins=[Ollama()])

@ai.flow()
async def summarize_notes(notes: str) -> str:
    response = await ai.generate(
        model='ollama/llama3.2',
        prompt=f'Summarize these release notes in three bullet points:\n\n{notes}',
        config=OllamaConfig(
            temperature=0.2,
            max_output_tokens=256,
            num_ctx=16_384,
            keep_alive='30m',
        ),
    )
    return response.text

For thinking-capable models (such as qwen3 or deepseek-r1), set think=True (or 'low', 'medium', 'high') on OllamaConfig. The model’s thinking comes back as reasoning parts on response.message.content, separate from response.text.

OllamaConfig rejects unknown keys, so a misspelled num_ctx fails instead of running with the default context window. max_output_tokens and stop_sequences map to Ollama’s num_predict and stop. Pass an Ollama option the class doesn’t declare in OllamaConfig(extra={...}).

Model capabilities

The plugin asks the server what each model supports instead of guessing from its name:

• Listing: the Dev UI’s model list calls GET /api/tags. Names containing embed become embedders; every other model gets one POST /api/show, at most four at a time, each bounded by five seconds or by timeout when that is shorter.
• Resolving by name: the first ai.generate(model='ollama/qwen3') makes the same /api/show call before the request.
• Mapping: the reported tools capability sets supports.tools, and vision sets supports.media.
• Caching: results are kept per process, keyed by model name and the digest from /api/tags, so a re-pulled model is probed again.
• Fallback: a failed probe (server down, model not pulled, a server too old to report capabilities) advertises every capability and is retried after 30 seconds. The request still goes to /api/chat.

Advanced usage

Embeddings

Any pulled embedding model, such as nomic-embed-text, resolves under 'ollama/<name>':

from genkit import Genkit
from genkit_ollama import Ollama

ai = Genkit(plugins=[Ollama()])

@ai.flow()
async def embed_text(text: str) -> list[float]:
    embeddings = await ai.embed(
        embedder='ollama/nomic-embed-text',
        content=text,
    )
    return embeddings[0].embedding

Streaming

from genkit import FlowContext

@ai.flow()
async def streaming_story(topic: str, ctx: FlowContext[str]) -> str:
    stream_response = ai.generate_stream(
        model='ollama/llama3.2',
        prompt=f'Write a short story about {topic}.',
    )
    async for chunk in stream_response.stream:
        ctx.send_chunk(chunk.text)
    return (await stream_response.response).text

Tool calling

Models with tool support (such as llama3.2, qwen3, and mistral-nemo) execute Genkit tools automatically:

from pydantic import BaseModel, Field

class WeatherInput(BaseModel):
    location: str = Field(description='City name')

@ai.tool()
async def get_weather(input: WeatherInput) -> str:
    """Get the current weather for a location."""
    return f'The weather in {input.location} is 72°F and sunny.'

@ai.flow()
async def weather_flow(location: str) -> str:
    response = await ai.generate(
        model='ollama/mistral-nemo',
        prompt=f"What's the weather like in {location}?",
        tools=[get_weather],
    )
    return response.text

Multimodal input

Pass images to vision-capable models (such as llava or gemma3) as data: or http(s): URLs via Part.from_media(). Local filesystem paths are rejected:

import base64
from pathlib import Path
from genkit import Part

image_b64 = base64.b64encode(Path('diagram.png').read_bytes()).decode('ascii')

response = await ai.generate(
    model='ollama/gemma3',
    prompt=[
        Part.from_text('Describe the components in this diagram.'),
        Part.from_media(url=f'data:image/png;base64,{image_b64}', content_type='image/png'),
    ],
)

Structured output

from pydantic import BaseModel, Field

class Recipe(BaseModel):
    name: str = Field(description='Recipe name')
    ingredients: list[str] = Field(description='List of ingredients')
    steps: list[str] = Field(description='Cooking steps')
    prep_time_minutes: int = Field(description='Preparation time in minutes')

@ai.flow()
async def create_recipe(dish: str) -> Recipe:
    response = await ai.generate(
        model='ollama/llama3.2',
        prompt=f'Create a recipe for {dish}',
        output_schema=Recipe,
    )
    return response.output