Writing
How to Write a Genkit Python Plugin from Scratch
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. Google ships the first-party ones (Google AI, Vertex AI, Anthropic, Ollama). The extension point exists 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 it as both a tool (callable by LLMs) and a flow (callable from your application). By the end you’ll understand the full plugin lifecycle and be ready to build your own.
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
from genkit._core._typing import ActionMetadata
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 actionsresolve() — 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 Nonelist_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}/weather-report",
action_type=ActionKind.FLOW,
description="Generate a full weather report",
),
]Plugin Configuration with Pydantic
Real plugins need configuration: API keys, base URLs, timeout settings. Use Pydantic for this—it gives you 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 = configUsage:
# 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, etc. 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
class GetWeatherInput(BaseModel):
city: str
units: str = "metric" # "metric" or "imperial"
class WeatherData(BaseModel):
city: str
temperature: float
condition: str
humidity: int
units: str
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.
Building the Weather Plugin: Flows
Flows are higher-level than tools. They’re user-facing entrypoints that can orchestrate multiple model calls and tool uses. From the LLM’s perspective, flows don’t exist—they’re callable from application code, not by the model.
A flow action uses ActionKind.FLOW:
from genkit import Genkit
from genkit._core._action import ActionRunContext, get_current_context
class WeatherPlugin(Plugin):
name = "weather"
def __init__(self, ai: Genkit, api_key: str, base_url: str = "https://api.weatherapi.com/v1"):
self._ai = ai # reference to the Genkit instance for generate() calls
self._api_key = api_key
self._base_url = base_url
def _make_weather_report_flow(self, get_weather_action: Action) -> Action:
"""Create the weather-report flow action."""
ai = self._ai
class WeatherReportInput(BaseModel):
city: str
include_forecast: bool = False
async def weather_report_fn(input: WeatherReportInput, ctx: ActionRunContext) -> str:
"""Generate a full weather report with analysis."""
# First, get the raw weather data
weather_data = await get_weather_action(
GetWeatherInput(city=input.city)
)
# Then, use LLM to generate a human-friendly report
prompt = f"""
Weather data for {input.city}: {weather_data}
Write a friendly 2-sentence weather report suitable for a general audience.
{"Include what to wear and activity suggestions." if input.include_forecast else ""}
"""
response = await ai.generate(prompt=prompt)
return response.text
return Action(
kind=ActionKind.FLOW,
name=f"{self.name}/weather-report",
fn=weather_report_fn,
description="Generate a human-friendly weather report for a city.",
)
async def init(self) -> list[Action]:
get_weather = self._make_get_weather_action()
weather_report = self._make_weather_report_flow(get_weather)
return [get_weather, weather_report]One architectural note: the flow needs a reference to the Genkit instance to call ai.generate(). This creates a circular dependency if you’re constructing both at the same time. The common solution is to pass the Genkit instance into the plugin constructor—the Genkit object is created first, then the plugin is initialized after.
Alternatively, you can use get_current_context() and ai.generate() from within the action fn closure, and pass ai at plugin construction time (before Genkit.__init__ is complete). The Genkit framework calls plugin.init() lazily during the first request, not synchronously during Genkit.__init__, so this works.
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.FLOW and local_name == "weather-report":
get_weather = self._make_get_weather_action()
return self._make_weather_report_flow(get_weather)
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=str(ActionKind.TOOL),
description="Get current weather conditions for a city.",
),
ActionMetadata(
name=f"{self.name}/weather-report",
action_type=str(ActionKind.FLOW),
description="Generate a human-friendly weather report.",
),
]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
Let’s look at GoogleAI for patterns worth borrowing. Its key structure (py/plugins/google-genai/src/genkit/plugins/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 NoneThree 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
...2. 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.
3. 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:
# pyproject.toml
[project]
dependencies = [
"genkit[google-genai] @ git+https://github.com/firebase/genkit.git#subdirectory=py/packages/genkit",
]# app.py
from genkit import Genkit
from genkit.plugins.google_genai import GoogleAI
from weather_plugin import WeatherPlugin, GetWeatherInput
ai = Genkit(
plugins=[
GoogleAI(),
WeatherPlugin(ai=..., api_key="your-weather-key"),
],
model="googleai/gemini-2.0-flash",
)
# Make the tool available to the LLM for interactive testing
@ai.flow()
async def weather_assistant(question: str) -> str:
# The weather plugin's tool is registered; the LLM can call it
response = await ai.generate(
prompt=question,
tools=["weather/get-weather"], # reference by name
)
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 tool and flow 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, production-ready. Note the architecture decision: the plugin registers tools only. Flows that orchestrate model calls are defined by application code, not by the plugin. This avoids the circular dependency of a plugin needing a reference to the Genkit instance that contains it.
# 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=str(ActionKind.TOOL),
description="Get current weather conditions for any city worldwide.",
),
ActionMetadata(
name=f"{self.name}/get-forecast",
action_type=str(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.plugins.google_genai import GoogleAI
from weather_plugin import WeatherPlugin
# Pass the plugin in the constructor
ai = Genkit(
plugins=[
GoogleAI(),
WeatherPlugin(api_key="your-weather-api-key"),
],
model="googleai/gemini-2.0-flash",
)
# Application-level flows use the plugin's registered tools
@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."""
# Call the tool action directly to get structured data
from weather_plugin import GetWeatherInput, GetForecastInput
# Tools are registered Actions; resolve and call them
weather_action = await ai.registry.resolve_action(
ActionKind.TOOL, "weather/get-weather"
)
current = await weather_action(GetWeatherInput(city=city))
response = await ai.generate(
prompt=f"Write a friendly 2-sentence weather report. Data: {current}"
)
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:
- ActionKind: MODEL · What it is: Language model (text/multimodal generation)
- ActionKind: EMBEDDER · What it is: Text embedding model
- ActionKind: TOOL · What it is: Callable by LLMs during generation
- ActionKind: FLOW · What it is: User-facing orchestration function
- ActionKind: EVALUATOR · What it is: Flow/model evaluation function
- ActionKind: RETRIEVER · What it is: Document retrieval for RAG
- ActionKind: INDEXER · What it is: Document indexing for RAG
- ActionKind: CUSTOM · What it is: 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 and flows, creating Action objects directly (as shown above) is the standard approach.
What to Expect as the SDK Matures
The plugin API is functional and used by all first-party plugins. But it’s not yet stable in the semver sense. A few things that may change before 1.0:
- Plugin construction pattern: Passing
aito the plugin constructor is a common need but not yet a first-class feature. Expect a cleaner API here. MiddlewarePluginsubclass: If your plugin only adds middleware (not models or tools), useMiddlewarePlugininstead ofPlugin—it removes the need to implementinit(),resolve(), andlist_actions().loop_local_client: This function is ingenkit.plugin_apiand stable, but the import path may be reorganized.
The plugin system is one of the most powerful and least documented parts of Genkit Python. Building on it now means building ahead of the documentation—but the source is clean, the interface is simple, and the patterns from first-party plugins give you everything you need.
Jeff Huang is Tech Lead for Genkit Python at Google. The Genkit Python SDK is open source at github.com/firebase/genkit.