Writing

How to Write a Genkit Python Plugin from Scratch

· updated

Zero guides exist on how to write one. This is that guide.

How to Write a Genkit Python Plugin from Scratch

The plugin system is how Genkit grows. Every model provider, every new action type, every cross-cutting middleware: it all flows through plugins. On PyPI, Google ships GoogleAI and VertexAI in the genkit-google-genai package. The same Plugin base class is the extension point for everyone else.

Zero guides exist on how to write one. This is that guide.

We’ll build a weather plugin from scratch: a Plugin subclass that wraps a REST API and exposes tools the model can call. Flows that orchestrate those tools stay in application code. By the end you’ll understand the plugin lifecycle and be ready to build your own.

This post matches PyPI genkit==0.8.1 and genkit-google-genai==0.8.1.

The Minimal Plugin Shell

Every Genkit plugin is a class that extends Plugin. The base class has three required methods:

from genkit.plugin_api import Action, ActionKind, ActionMetadata, Plugin

class MyPlugin(Plugin):
    name = "myplugin"  # namespace prefix for all actions this plugin registers

    async def init(self) -> list[Action]:
        """Called once at startup. Return actions to pre-register."""
        return []

    async def resolve(self, action_type: ActionKind, name: str) -> Action | None:
        """Called lazily when an action is needed. Return None if not found."""
        return None

    async def list_actions(self) -> list[ActionMetadata]:
        """Return action metadata for the Dev UI."""
        return []

That’s it. That’s a valid plugin. Pass it to Genkit:

from genkit import Genkit

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

Nothing useful happens yet, but the framework accepts it, initializes it, and queries it for actions when needed. Let’s fill it in.

Understanding the Three Methods

Before building the weather plugin, you need a clear mental model of when each method is called.

init(): Startup Registration

init() runs once, when the Genkit runtime initializes your plugin. Use it to pre-register actions you know exist at startup, especially when you need to make an API call to discover what’s available (like how GoogleAI calls client.models.list() during init).

Return a list of Action objects. Genkit registers each one under your plugin’s namespace.

async def init(self) -> list[Action]:
    # Fetch what's available at startup
    actions = []
    for endpoint in await self._discover_endpoints():
        action = Action(
            kind=ActionKind.TOOL,
            name=f"{self.name}/{endpoint.name}",
            fn=self._make_handler(endpoint),
            description=endpoint.description,
        )
        actions.append(action)
    return actions

resolve(): Lazy Action Creation

resolve() is called when your application requests an action that isn’t already in the registry. This is how GoogleAI handles model requests: it doesn’t pre-register every model at startup; it creates model Actions on demand.

Return None if you can’t resolve the requested action. Return an Action if you can.

async def resolve(self, action_type: ActionKind, name: str) -> Action | None:
    if action_type == ActionKind.MODEL and name.startswith(f"{self.name}/"):
        model_name = name[len(f"{self.name}/"):]
        return self._create_model_action(model_name)
    return None

list_actions(): Dev UI Discovery

list_actions() returns metadata about all actions your plugin can provide. This is what populates the Dev UI’s action browser. It doesn’t need to instantiate the actions, just describe them.

async def list_actions(self) -> list[ActionMetadata]:
    return [
        ActionMetadata(
            name=f"{self.name}/get-weather",
            action_type=ActionKind.TOOL,
            description="Get current weather for a city",
        ),
        ActionMetadata(
            name=f"{self.name}/get-forecast",
            action_type=ActionKind.TOOL,
            description="Get a multi-day weather forecast",
        ),
    ]

Plugin Configuration with Pydantic

Real plugins need configuration: API keys, base URLs, timeout settings. Use Pydantic for this. You get validation, default values, and clear error messages.

from pydantic import BaseModel, Field

class WeatherPluginConfig(BaseModel):
    """Configuration for the Weather plugin."""
    api_key: str = Field(description="API key for weather service")
    base_url: str = Field(
        default="https://api.weatherapi.com/v1",
        description="Base URL for weather API"
    )
    timeout_seconds: float = Field(default=10.0, ge=0.1, le=60.0)
    default_units: str = Field(default="metric", pattern="^(metric|imperial)$")

class WeatherPlugin(Plugin):
    name = "weather"

    def __init__(self, config: WeatherPluginConfig | None = None, **kwargs):
        """Accept config as Pydantic model or keyword args."""
        if config is None:
            config = WeatherPluginConfig(**kwargs)
        self._config = config

Usage:

# Either way works
ai = Genkit(plugins=[
    WeatherPlugin(api_key="your-key"),
])

# Or explicit config object
ai = Genkit(plugins=[
    WeatherPlugin(config=WeatherPluginConfig(
        api_key="your-key",
        default_units="imperial",
    ))
])

This is the pattern first-party plugins use. GoogleAI.__init__ takes api_key, credentials, http_options, and related options, and stores them on the instance. The plugin constructor is your only chance to capture configuration before init() is called.

Building the Weather Plugin: Tools

Tools in Genkit are Action objects with ActionKind.TOOL. They need typed inputs (Pydantic BaseModel) and a string-compatible return value. The LLM calls them by name; Genkit handles the schema generation and function dispatch.

Here’s the tool implementation for the weather plugin:

import httpx
from pydantic import BaseModel
from genkit.plugin_api import Action, ActionKind, Plugin

class GetWeatherInput(BaseModel):
    city: str
    units: str = "metric"  # "metric" or "imperial"

class WeatherPlugin(Plugin):
    name = "weather"

    def __init__(self, api_key: str, base_url: str = "https://api.weatherapi.com/v1"):
        self._api_key = api_key
        self._base_url = base_url

    def _make_get_weather_action(self) -> Action:
        """Create the get-weather tool action."""
        api_key = self._api_key
        base_url = self._base_url

        async def get_weather_fn(input: GetWeatherInput) -> str:
            """Get current weather for a city."""
            async with httpx.AsyncClient() as client:
                response = await client.get(
                    f"{base_url}/current.json",
                    params={
                        "key": api_key,
                        "q": input.city,
                        "aqi": "no",
                    },
                    timeout=10.0,
                )
                response.raise_for_status()
                data = response.json()

                temp_key = "temp_c" if input.units == "metric" else "temp_f"
                unit_label = "°C" if input.units == "metric" else "°F"

                current = data["current"]
                location = data["location"]

                return (
                    f"{location['name']}: {current[temp_key]}{unit_label}, "
                    f"{current['condition']['text']}, "
                    f"humidity {current['humidity']}%"
                )

        return Action(
            kind=ActionKind.TOOL,
            name=f"{self.name}/get-weather",
            fn=get_weather_fn,
            description="Get current weather conditions for a city.",
        )

    async def init(self) -> list[Action]:
        return [
            self._make_get_weather_action(),
        ]

A few things to notice:

The fn closure captures config. api_key and base_url are captured in the closure, not stored on the action object. This is idiomatic Python: actions are stateless functions from Genkit’s perspective.

The function signature drives schema generation. Genkit introspects get_weather_fn’s type annotations and builds a JSON schema for the LLM automatically. The input must be a BaseModel. Bare primitives (city: str) cause Gemini to reject the schema with a 400 error.

The return type is str. Tools can return strings directly. Genkit serializes the result for the LLM. You can also return Pydantic models or dicts.

Keep Flows in Application Code

Plugins that wrap an external API should usually register tools (and, for providers, models or embedders). Flows that call ai.generate() belong in the app that owns the Genkit instance.

If you put a flow Action inside the plugin, the flow needs a reference to ai so it can call generate(). That creates a circular construction problem: the plugin is an argument to Genkit(...), so it cannot cleanly hold the finished Genkit object. The tools-only shape avoids that entirely. Your app defines @ai.flow() functions that pass tools=["weather/get-weather"] into generate().

How resolve() and list_actions() Complete the Picture

The init() approach pre-registers everything at startup. That’s fine for a small, fixed set of actions. If your plugin wraps a dynamic API with hundreds of endpoints (like GoogleAI with its model catalog), use resolve() instead:

async def resolve(self, action_type: ActionKind, name: str) -> Action | None:
    """Lazily create actions by name, on demand."""
    prefix = f"{self.name}/"
    if not name.startswith(prefix):
        return None

    local_name = name[len(prefix):]

    if action_type == ActionKind.TOOL and local_name == "get-weather":
        return self._make_get_weather_action()

    if action_type == ActionKind.TOOL and local_name == "get-forecast":
        return self._make_get_forecast_action()

    return None

async def list_actions(self) -> list[ActionMetadata]:
    """Advertise actions to the Dev UI without instantiating them."""
    return [
        ActionMetadata(
            name=f"{self.name}/get-weather",
            action_type=ActionKind.TOOL,
            description="Get current weather conditions for a city.",
        ),
        ActionMetadata(
            name=f"{self.name}/get-forecast",
            action_type=ActionKind.TOOL,
            description="Get a multi-day weather forecast.",
        ),
    ]

The Dev UI calls list_actions() when it renders the action browser. resolve() is called when your application actually invokes an action. If you only implement init(), the Dev UI shows what’s registered; if you implement resolve() and list_actions(), the Dev UI shows everything your plugin can provide, whether it’s been used yet or not.

How First-Party Plugins Do It

Look at GoogleAI for patterns worth borrowing. On PyPI it lives in the genkit-google-genai package (genkit_google_genai/google.py in the installed layout; in the monorepo, py/packages/genkit-google-genai/src/genkit_google_genai/google.py):

class GoogleAI(Plugin):
    name = "googleai"    # all models namespaced as "googleai/..."
    _vertexai = False

    def __init__(self, api_key=None, ...):
        # Store config at construction time
        self._client_kwargs = {"api_key": api_key, ...}
        # Use loop_local_client for thread-safe client creation
        self._runtime_client = loop_local_client(
            lambda: genai.client.Client(**self._client_kwargs)
        )

    async def init(self) -> list[Action]:
        # Discover models from API
        genai_models = _list_genai_models(self._runtime_client(), is_vertex=False)
        actions = []
        for name in genai_models.gemini:
            actions.append(self._resolve_model(f"googleai/{name}"))
        return actions

    async def resolve(self, action_type, name) -> Action | None:
        # On-demand action creation for unknown model names
        if action_type == ActionKind.MODEL:
            return self._resolve_model(name)
        return None

Three patterns to borrow:

  1. loop_local_client for SDK clients. If you’re wrapping an SDK that creates event loop-bound resources (HTTP clients, gRPC channels), use loop_local_client from genkit.plugin_api. It creates one instance per event loop, which matters in test environments where multiple loops are created and destroyed.
from genkit.plugin_api import loop_local_client
import httpx

class MyPlugin(Plugin):
    def __init__(self, api_key: str):
        self._client = loop_local_client(
            lambda: httpx.AsyncClient(
                headers={"X-API-Key": api_key},
                timeout=10.0
            )
        )

    async def init(self) -> list[Action]:
        client = self._client()  # get loop-local instance
        ...

  1. Separate _resolve_* helpers. GoogleAI has _resolve_model(), _resolve_embedder(), _resolve_veo_model(). Each returns one Action. Both init() and resolve() call these helpers. This avoids duplicating action creation logic.

  2. name class variable, not instance variable. The plugin name is a class-level string. Genkit uses it as the namespace prefix: all actions returned from init() have their names normalized to ensure they start with {plugin.name}/. You don’t have to manually prefix every action name, but being explicit (using f"{self.name}/my-action") makes the code clearer.

Testing with the Dev UI

The Dev UI (genkit start) auto-discovers your plugin’s actions via list_actions() and renders them in the action browser. Running locally:

uv add genkit==0.8.1 genkit-google-genai==0.8.1
# app.py
from genkit import Genkit
from genkit_google_genai import GoogleAI
from weather_plugin import WeatherPlugin

ai = Genkit(
    plugins=[
        GoogleAI(),
        WeatherPlugin(api_key="your-weather-key"),
    ],
    model="googleai/gemini-2.0-flash",
)

# Application flow: the LLM can call the plugin's registered tools
@ai.flow()
async def weather_assistant(question: str) -> str:
    response = await ai.generate(
        prompt=question,
        tools=["weather/get-weather"],
    )
    return response.text

if __name__ == "__main__":
    ai.run_main(main())

Run genkit start -- python app.py and open the Dev UI. You’ll see your plugin’s tools in the action browser. You can invoke them directly from the UI to test inputs, inspect outputs, and view traces, without writing test harnesses.

Complete Example: The Weather Plugin

Here’s the full weather plugin. It registers tools only. Flows that orchestrate model calls are defined by application code, not by the plugin.

# weather_plugin.py

import httpx
from pydantic import BaseModel, Field
from genkit.plugin_api import Action, ActionKind, ActionMetadata, loop_local_client, Plugin

class WeatherPluginConfig(BaseModel):
    api_key: str
    base_url: str = "https://api.weatherapi.com/v1"
    timeout: float = Field(default=10.0, ge=0.1, le=60.0)

class GetWeatherInput(BaseModel):
    city: str
    units: str = Field(default="metric", pattern="^(metric|imperial)$")

class GetForecastInput(BaseModel):
    city: str
    days: int = Field(default=3, ge=1, le=7)

class WeatherPlugin(Plugin):
    name = "weather"

    def __init__(self, config: WeatherPluginConfig | None = None, **kwargs):
        if config is None:
            config = WeatherPluginConfig(**kwargs)
        self._config = config
        # Thread-safe, loop-local HTTP client
        self._http = loop_local_client(
            lambda: httpx.AsyncClient(
                base_url=config.base_url,
                timeout=config.timeout,
            )
        )

    def _make_get_weather_action(self) -> Action:
        config = self._config
        http = self._http

        async def get_weather(input: GetWeatherInput) -> str:
            """Get current weather conditions for a city."""
            client = http()
            response = await client.get(
                "/current.json",
                params={"key": config.api_key, "q": input.city, "aqi": "no"},
            )
            response.raise_for_status()
            data = response.json()

            temp_key = "temp_c" if input.units == "metric" else "temp_f"
            unit = "°C" if input.units == "metric" else "°F"
            loc = data["location"]["name"]
            cur = data["current"]

            return (
                f"{loc}: {cur[temp_key]}{unit}, "
                f"{cur['condition']['text']}, "
                f"humidity {cur['humidity']}%"
            )

        return Action(
            kind=ActionKind.TOOL,
            name=f"{self.name}/get-weather",
            fn=get_weather,
            description="Get current weather conditions for any city worldwide.",
        )

    def _make_get_forecast_action(self) -> Action:
        config = self._config
        http = self._http

        async def get_forecast(input: GetForecastInput) -> str:
            """Get multi-day weather forecast for a city."""
            client = http()
            response = await client.get(
                "/forecast.json",
                params={
                    "key": config.api_key,
                    "q": input.city,
                    "days": input.days,
                },
            )
            response.raise_for_status()
            data = response.json()

            days = data["forecast"]["forecastday"]
            lines = []
            for day in days:
                d = day["day"]
                lines.append(
                    f"{day['date']}: {d['avgtemp_c']}°C avg, "
                    f"{d['condition']['text']}"
                )
            return "\n".join(lines)

        return Action(
            kind=ActionKind.TOOL,
            name=f"{self.name}/get-forecast",
            fn=get_forecast,
            description="Get a multi-day weather forecast for any city.",
        )

    async def init(self) -> list[Action]:
        return [
            self._make_get_weather_action(),
            self._make_get_forecast_action(),
        ]

    async def resolve(self, action_type: ActionKind, name: str) -> Action | None:
        local = name.removeprefix(f"{self.name}/")
        if action_type == ActionKind.TOOL:
            if local == "get-weather":
                return self._make_get_weather_action()
            if local == "get-forecast":
                return self._make_get_forecast_action()
        return None

    async def list_actions(self) -> list[ActionMetadata]:
        return [
            ActionMetadata(
                name=f"{self.name}/get-weather",
                action_type=ActionKind.TOOL,
                description="Get current weather conditions for any city worldwide.",
            ),
            ActionMetadata(
                name=f"{self.name}/get-forecast",
                action_type=ActionKind.TOOL,
                description="Get a multi-day weather forecast for any city.",
            ),
        ]

Using it in an application:

# app.py

from genkit import Genkit
from genkit_google_genai import GoogleAI
from weather_plugin import WeatherPlugin

ai = Genkit(
    plugins=[
        GoogleAI(),
        WeatherPlugin(api_key="your-weather-api-key"),
    ],
    model="googleai/gemini-2.0-flash",
)

@ai.flow()
async def travel_advisor(destination: str) -> str:
    """Should I visit this city this weekend? Uses real weather data."""
    response = await ai.generate(
        prompt=f"Should I visit {destination} this weekend? Consider the weather.",
        tools=["weather/get-weather", "weather/get-forecast"],
    )
    return response.text

@ai.flow()
async def weather_report(city: str) -> str:
    """Full weather report with analysis."""
    response = await ai.generate(
        prompt=(
            f"Write a friendly 2-sentence weather report for {city}. "
            "Call the weather tools for current conditions and forecast."
        ),
        tools=["weather/get-weather", "weather/get-forecast"],
    )
    return response.text

async def main():
    result = await travel_advisor("Chicago")
    print(result)

    report = await weather_report("Tokyo")
    print(report)

if __name__ == "__main__":
    ai.run_main(main())

What Plugins Can Register

The ActionKind enum shows everything the registry understands. Common ones for a plugin:

  • MODEL: language model (text/multimodal generation)
  • EMBEDDER: text embedding model
  • TOOL: callable by LLMs during generation
  • FLOW: user-facing orchestration function (usually defined with @ai.flow() in app code)
  • EVALUATOR: flow/model evaluation function
  • RETRIEVER / INDEXER: document retrieval and indexing for RAG
  • BACKGROUND_MODEL: long-running operations (start/check/cancel)
  • CUSTOM: arbitrary registered function

Your plugin can register any of these. The pattern is the same: create an Action with the appropriate kind and a function that matches the expected signature for that kind.

For model actions specifically, use ai.define_model() rather than creating Action objects directly. It handles model-specific schema setup and wraps the response type correctly. For tools, creating Action objects directly (as shown above) is the standard approach inside a plugin.

What to Expect as the SDK Matures

The plugin API is functional and used by the first-party Google plugins. A few things to keep in mind on 0.8.1:

  • Prefer genkit.plugin_api for plugin authoring (Plugin, Action, ActionKind, ActionMetadata, loop_local_client, MiddlewarePlugin). Avoid importing private _core modules in application or plugin code.
  • If your plugin only adds middleware (not models or tools), use MiddlewarePlugin instead of Plugin. It removes the need to implement init(), resolve(), and list_actions().
  • loop_local_client is in genkit.plugin_api and is the supported way to keep SDK clients loop-local.

The plugin system is one of the least documented parts of Genkit Python. Building on it now means reading the source and the first-party plugins. The interface is small, and the patterns above are enough to ship a useful plugin.

Jeff Huang is Tech Lead for Genkit Python at Google. The Genkit Python SDK is open source at github.com/genkit-ai/genkit.