How to Build an AI Agent for Meeting Notes

def get_transcript(meeting, asr, seen: set) -> dict | None:
    if not meeting["consent_to_record"]:        # no consent, no processing
        return None
    if meeting["id"] in seen:                    # already transcribed
        return None
    seen.add(meeting["id"])
    text = asr.transcribe(meeting["recording_url"],   # Whisper / AssemblyAI / Deepgram
                          diarize=True)               # speaker labels + timestamps
    return {"id": meeting["id"], "title": meeting["title"],
            "attendees": meeting["attendees"], "text": text}
# Meeting id makes it idempotent: each call is handled once, even on a re-run.

from google import genai
client = genai.Client()

def summarize(transcript: str) -> str:
    return client.models.generate_content(
        model="gemini-2.5-flash",
        contents=[f"Summarize this meeting transcript in 4 to 6 sentences. "
                  f"Use only what is said; do not infer.\n{transcript}"]).text

from google import genai
from google.genai import types
from pydantic import BaseModel
client = genai.Client()

class Action(BaseModel):
    what: str; owner: str | None = None; due: str | None = None

class Notes(BaseModel):
    decisions: list[str]; actions: list[Action]

def extract(transcript: str) -> Notes:
    res = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=["Extract decisions and action items. Use only what is "
                  "stated. Leave owner or due null if not said.\n" + transcript],
        config=types.GenerateContentConfig(
            response_mime_type="application/json", response_schema=Notes))
    return res.parsed                # a typed Notes object

def match_owner(name: str | None, directory: list[dict]) -> dict | None:
    if not name:
        return None                              # no name was said
    hits = [p for p in directory
            if name.lower() in p["name"].lower()]
    if len(hits) == 1:
        return hits[0]                           # one clear match
    return None                                  # 0 or many -> a person assigns
# A simple, auditable lookup; an unresolved owner is left for review, not guessed.

def create_items(meeting: dict, notes, directory: list[dict]) -> None:
    for a in notes.actions:
        owner = match_owner(a.owner, directory)
        tasks_api.create(title=a.what, due=a.due,
                         assignee=owner["id"] if owner else None,
                         source=meeting["id"])    # linked to the recording
    audit.log("items_created", meeting["id"], notes.actions)
# Tasks land in the tools you already use, traceable to the source call.

def queue_notes(meeting: dict, summary: str, notes) -> None:
    approval_queue.add(meeting=meeting["id"], summary=summary,
                       decisions=notes.decisions, actions=notes.actions)

def on_approval(meeting: dict, final_notes: dict) -> None:
    distributor.send(to=meeting["attendees"], body=final_notes)   # email / chat
    audit.log("notes_sent", meeting["id"], final_notes)   # only after approval
# The agent drafts and queues; distribution happens only when a person approves.

def store_notes(meeting: dict, summary: str, notes) -> None:
    notes_store.put(meeting_id=meeting["id"], title=meeting["title"],
                    summary=summary, decisions=notes.decisions,
                    actions=notes.actions,
                    recording=meeting["recording_url"],   # trace to the source
                    transcript_ref=meeting["id"])
    search_index.add(meeting["id"], summary)   # findable across past calls
# Every note links back to the recording it came from, and is searchable later.

from google.adk.agents import SequentialAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

notes_pipeline = SequentialAgent(name="notes_pipeline",
    sub_agents=[summarizer, extractor, matcher])   # each an LlmAgent with tools
runner = Runner(agent=notes_pipeline, app_name="notes",
                session_service=InMemorySessionService())

def guard(meeting: dict) -> str | None:
    if not meeting["consent_to_record"]:  return "skip"          # consent first
    if store.seen(meeting["id"]):         return "already_done"  # idempotent
    return None
# Run on each new recording. Notes are distributed only after a person approves.