Developing Genkit Python

uv add genkit-fastapi fastapi uvicorn

import uvicorn
from fastapi import FastAPI
from genkit import Genkit
from genkit.agent import InMemorySessionStore
from genkit_fastapi import serve_agent, serve_flow
from genkit_google_genai import GoogleAI

ai = Genkit(plugins=[GoogleAI()], model='googleai/gemini-flash-latest')

agent = ai.define_agent(
    name='weatherAgent',
    system='Weather assistant. Be concise.',
    store=InMemorySessionStore(),
)

@ai.flow()
async def hello(name: str) -> str:
    return f'Hello, {name}'

app = FastAPI()
# Routes use the action name as defined: agent `weatherAgent` → /api/weatherAgent;
# flow `hello` → /api/hello. Override with base_path=... if needed.
app.include_router(serve_agent(agent), prefix='/api')
app.include_router(serve_flow(hello), prefix='/api')

# Under genkit start, drive uvicorn from ai.run_main (don't nest uvicorn.run inside it).
async def main() -> None:
    import os
    config = uvicorn.Config(app, host='0.0.0.0', port=int(os.environ.get('PORT', '8080')))
    await uvicorn.Server(config).serve()

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

data: {"message": "<chunk text>"}   ← one per ctx.send_chunk() call
data: {"message": "<chunk text>"}
data: {"result": <final output>}    ← sent once when the action completes

const res = await fetch('/api/chat', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json', 'Accept': 'text/event-stream' },
  body: JSON.stringify({ data: { topic: 'quantum computing' } }),
});
const reader = res.body.getReader();
// decode and parse each `data: {...}` line

curl -N -X POST http://localhost:8080/api/chat \
  -H 'Content-Type: application/json' \
  -H 'Accept: text/event-stream' \
  -d '{"data": {"topic": "quantum computing"}}'

import uvicorn
from pydantic import BaseModel
from fastapi import FastAPI
from genkit import Genkit, ActionRunContext
from genkit_fastapi import serve_flow
from genkit_google_genai import GoogleAI

ai = Genkit(plugins=[GoogleAI()], model='googleai/gemini-flash-latest')


class ChatInput(BaseModel):
    topic: str


@ai.flow()
async def chat(input: ChatInput, ctx: ActionRunContext) -> str:
    sr = ai.generate_stream(prompt=f'Tell me about {input.topic}.')
    full = ''
    async for chunk in sr.stream:
        if chunk.text:
            ctx.send_chunk(chunk.text)   # each chunk → SSE event on the wire
            full += chunk.text
    return full


app = FastAPI()
app.include_router(serve_flow(chat), prefix='/api')  # POST /api/chat

async def main() -> None:
    import os
    config = uvicorn.Config(app, host='0.0.0.0', port=int(os.environ.get('PORT', '8080')))
    await uvicorn.Server(config).serve()

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

class ResearchInput(BaseModel):
    topic: str

@ai.flow()
async def research(input: ResearchInput, ctx: ActionRunContext) -> str:
    sr = ai.generate_stream(prompt=f'Explain {input.topic} in depth.')
    full = ''
    async for chunk in sr.stream:
        if chunk.text:
            ctx.send_chunk(chunk.text)
            full += chunk.text
    return full


class HeadlineInput(BaseModel):
    text: str

@ai.flow()
async def make_headline(input: HeadlineInput) -> str:
    response = await ai.generate(prompt=f'One-line headline for: {input.text}')
    return response.text.strip()


class ReportInput(BaseModel):
    topic: str

@ai.flow()
async def report(input: ReportInput, ctx: ActionRunContext) -> str:
    headline = await make_headline(HeadlineInput(text=input.topic))
    ctx.send_chunk(f'# {headline}\n\n')

    body = (
        await research.run(
            ResearchInput(topic=input.topic),
            on_chunk=ctx.send_chunk,
        )
    ).response

    return f'# {headline}\n\n{body}'


app.include_router(serve_flow(report), prefix='/api')  # POST /api/report

import asyncio

class AnalysisInput(BaseModel):
    text: str

class CheckResult(BaseModel):
    issues: list[str]

class CombinedAnalysis(BaseModel):
    issues: list[str]

@ai.flow()
async def check_security(input: AnalysisInput) -> CheckResult:
    r = await ai.generate(
        prompt=f'List security concerns as a short comma-separated line (or "none"): {input.text[:2000]}',
    )
    raw = (r.text or '').strip()
    issues = [s.strip() for s in raw.split(',') if s.strip() and s.strip().lower() != 'none']
    return CheckResult(issues=issues)

@ai.flow()
async def check_bugs(input: AnalysisInput) -> CheckResult:
    r = await ai.generate(
        prompt=f'List likely bugs or correctness issues as a short comma-separated line (or "none"): {input.text[:2000]}',
    )
    raw = (r.text or '').strip()
    issues = [s.strip() for s in raw.split(',') if s.strip() and s.strip().lower() != 'none']
    return CheckResult(issues=issues)

@ai.flow()
async def check_style(input: AnalysisInput) -> CheckResult:
    r = await ai.generate(
        prompt=f'List style or clarity issues as a short comma-separated line (or "none"): {input.text[:2000]}',
    )
    raw = (r.text or '').strip()
    issues = [s.strip() for s in raw.split(',') if s.strip() and s.strip().lower() != 'none']
    return CheckResult(issues=issues)

@ai.flow()
async def analyze(input: AnalysisInput) -> CombinedAnalysis:
    security, bugs, style = await asyncio.gather(
        check_security(input),
        check_bugs(input),
        check_style(input),
    )
    return CombinedAnalysis(issues=security.issues + bugs.issues + style.issues)


app.include_router(serve_flow(analyze), prefix='/api')  # POST /api/analyze

class SentimentResult(BaseModel):
    sentiment: str        # positive / negative / neutral
    confidence: float     # 0.0–1.0
    key_phrases: list[str]

@ai.flow()
async def sentiment(input: AnalysisInput) -> SentimentResult:
    response = await ai.generate(
        prompt=f'Analyze sentiment: {input.text}',
        output_format='json',
        output_schema=SentimentResult,
    )
    return response.output


app.include_router(serve_flow(sentiment), prefix='/api')  # POST /api/sentiment

from fastapi import Depends, Request
from genkit_fastapi import handle_genkit_request


async def current_user(request: Request) -> dict:
    # Resolve auth however you like.
    return {'uid': request.headers.get('x-user-id', 'anon')}


@app.post('/api/secure-chat')
async def secure_chat(
    request: Request,
    user: dict = Depends(current_user),
):
    return await handle_genkit_request(
        request,
        action=chat,
        context={'auth': user},
    )

GEMINI_API_KEY=your-key genkit start -- uv run src/main.py

Genkit Developer UI: http://localhost:4000

import httpx, asyncio

async with httpx.AsyncClient(base_url='http://127.0.0.1:18524') as client:
    async def call(path, data):
        r = await client.post(path, json={'data': data})
        r.raise_for_status()
        return r.json()['result']
    a, b = await asyncio.gather(
        call('/api/summarize', {'text': '...'}),
        call('/api/sentiment', {'text': '...'}),
    )