Announcing CIMD support for MCP Client registration
Learn more

Build a Performance Review Agent: Airtable, Google Forms, Notion, Slack

Shri Mithran
Director of Marketing

TL;DR

  • The agent collects structured ratings from Airtable and free-text feedback from Google Forms, groups both by employee, writes a per-employee summary page to Notion, and DMs the manager a Slack digest with links. One cycle per manager, then it exits.
  • Review data is the most sensitive payload in any of these agent templates, and the template scopes it in application code, not at the credential layer. resolve_direct_reports() filters Airtable records by a Manager Email column. That is a correctness convention. The Airtable connected account can still read every row in the table. This post shows where the boundary actually sits and what it takes to move it.
  • Two of the four connectors are MCP variants with different tool names, different parameter names, and a different response envelope than their REST siblings. Page creation exists only on Notion MCP; the plain Notion connector does not expose it.
  • Clone the repo, configure four connections, and run one cycle in under 30 minutes. The performance review collector template breaks the same pipeline into six copyable steps.

Review season arrives and the manager's job is not judgment. It is retrieval. Ratings live in an Airtable table, written notes live in a Google Form, and the two only line up if someone types the same employee name in both places. A manager with six reports spends a weekend copy-pasting before writing a single sentence of actual assessment.

So you automate it. An agent reads Airtable, reads the Form, groups by employee, writes a summary page in Notion, and DMs the manager a digest. Two hundred lines, one afternoon.

Then someone asks a question you do not have a good answer to: whose credential is reading that table?

Because the table has every employee in it. Not just this manager's six. If the agent authenticates as a shared HR bot, the manager's digest is correct only because your application code filtered it correctly. One wrong WHERE clause and a manager reads their peer's feedback about their own team. Nothing errors. The page renders. It is simply wrong in a way that is very difficult to walk back.

Where the Scope Boundary Actually Sits

This agent has one property none of the other templates have: it reads data about people who are not the person running it. That changes what "correct" means.

Here is the scoping logic, verbatim from aggregator.py:

def resolve_direct_reports( airtable_records: List[Dict], manager_field: str, employee_field: str, manager_email: str, configured_reports: Optional[List[str]], ) -> List[str]: """ Determine which employees are in scope for this manager. Prefers the Airtable Manager field (source of truth per review record). Falls back to the configured DIRECT_REPORTS list if no record names the manager. """ from_airtable = {} for record in airtable_records: fields = record.get("fields", {}) record_manager = str(fields.get(manager_field, "")).strip().lower() if record_manager == manager_email.strip().lower(): employee = fields.get(employee_field) if employee: normalized = str(employee).strip().casefold() from_airtable.setdefault(normalized, str(employee).strip()) if from_airtable: return sorted(from_airtable.values()) if configured_reports: logger.warning( f"No Airtable records tagged with manager '{manager_email}' " f"falling back to DIRECT_REPORTS env list" ) return configured_reports logger.warning(f"No direct reports found for manager '{manager_email}' in Airtable or config") return []

Read the order of operations. list_all_records() has already run. Every review record in the base is in memory, for every employee, under every manager. The filter happens after.

That is a fine design for a single-manager script you run yourself. It is the wrong design the moment this becomes a People Ops tool that several managers trigger, because the isolation is a == comparison in Python rather than a property of the credential. A misconfigured database query leaks data passively. A misconfigured agent acts actively: it writes the wrong summary to a Notion page and DMs it to the wrong manager, and now the leak has a timestamp and a reader.

Three ways to move the boundary, in increasing order of cost:

Filter at the connector, not in memory. airtable_list_records accepts a filter formula and a view parameter. Point AIRTABLE_VIEW at a per-manager view, or pass a formula constraining Manager Email, and the rows never reach the process. This is cheap and it is the first thing to do. It is still not isolation, because the credential could read the other rows if asked.

Scope the Airtable grant per manager. Airtable OAuth is granted per base. If each manager authorizes their own Airtable connected account against a base containing only their reports, then what the manager cannot read, the agent cannot read. This is real isolation, and it costs you a base-per-team data model.

Put approval in front of the write. The Notion write and the Slack DM are the two irreversible steps. Gating them behind a confirmation is the pattern described in access control for multi-tenant AI agents, and for review data it is worth the friction on the first few cycles.

The template ships the first version because it is the one you can run today. Be clear-eyed about which one you are deploying. There is a fuller treatment in how to implement least privilege for AI agent tool calls.

How the Collector Works: A Deterministic Pipeline

No reasoning loop, no memory, no decisions about what to fetch next. The agent runs a fixed sequence for one manager and one review period, then exits.

  1. Auth check. Confirm all four connected accounts are ACTIVE; surface a magic link for any that are not.
  2. Provisioning. Create the Airtable review table if it is missing. Validate the Google Form is reachable and has questions.
  3. Fetch. Paginate Airtable records to the end, paginate Google Forms responses to the end.
  4. Aggregate and summarize. Resolve direct reports, group ratings and comments per employee, write a narrative summary.
  5. Write. Upsert one Notion page per employee, matched by title.
  6. Notify. DM the manager one digest with per-employee counts, averages, and page links.

Two design decisions in there are worth naming because they are the ones that make re-runs safe.

The Notion write is an upsert, not a create. Running the same cycle twice updates the page in place instead of producing a second "Alex Kim, Q2 2026". The Slack DM is guarded by a processed-cycle state file keyed on (manager_email, review_period), so a re-run does not re-notify. The manager gets pinged once per period, no matter how many times the cron fires.

The Tool Surface: 83 Tools, Nine Calls

Four connectors is four catalogs, and this pipeline touches a thin slice of each.

Connector
Auth
Tools in the catalog
Tools this agent calls
Airtable
OAuth 2.0
26
airtable_get_base_schema, airtable_create_table, airtable_list_records
Google Forms
OAuth 2.0
10
googleforms_get_form, googleforms_list_responses
Notion MCP
OAuth 2.1 with DCR
28
notionmcp_notion-search, notionmcp_notion-create-pages, notionmcp_notion-update-page
Slack MCP
OAuth 2.1
19
slackmcp_slack_send_message

83 tools available, nine invoked. For an agent that reads performance feedback, the size of that gap is not a token-efficiency footnote, it is a containment property. Two of the three Airtable tools this agent calls are schema-level (airtable_get_base_schema needs schema.bases:read, airtable_create_table needs schema.bases:write), which means the connected account is already carrying write access to the base structure. Handing that same account a full 26-tool surface puts record deletion one bad tool selection away from an employee's review history.

The agent receives only the tools the current connected account is authorized to call. Not the catalog. If you want that enforced at the endpoint rather than by convention in your code, a Virtual MCP server declares the allowed tool list per agent role and is worth reading before this goes anywhere near production HR data.

Two Connector Variants, Two Payload Shapes

This is the detail that costs people an afternoon, so it goes before the setup steps rather than in troubleshooting.

Scalekit ships two variants of both Notion and Slack, and they are not interchangeable.

Plain REST connector
MCP connector
Notion
51 tools, OAuth 2.0
28 tools, OAuth 2.1 with DCR
Slack
91 tools, OAuth 2.0
19 tools, OAuth 2.1
Send-message signature
slack_send_message(channel, text)
slackmcp_slack_send_message(channel_id, message)
Response shape
flat payload dict
{"content": [{"type": "text", "text": "..."}]}

You need the Notion MCP variant. The page-creation tools live there. The plain Notion connector has almost twice the tool count and still does not expose notion-create-pages. Picking the bigger catalog is the wrong instinct here.

Slack works on either, with different parameter names. The repo detects which one you configured by substring match rather than an exact name lookup, because connection names are workspace-specific:

class SlackConnector(Connector): """ Scalekit exposes two Slack connector variants with different send-message tool signatures: - "*mcp*" variant -> slackmcp_slack_send_message(channel_id=..., message=...) - anything else -> slack_send_message(channel=..., text=...) Connection names are workspace-specific (e.g. "slackmcp", "slack-sKfekCVz"), so the MCP-vs-plain distinction is detected by substring match on "mcp". """ def __init__(self, actions, identifier: str, connector_name: str = "slackmcp"): super().__init__(actions, connector_name, identifier) if "mcp" in connector_name.lower(): self._send_tool = "slackmcp_slack_send_message" self._channel_param = "channel_id" self._text_param = "message" else: self._send_tool = "slack_send_message" self._channel_param = "channel" self._text_param = "text" def send_dm(self, user_id: str, text: str) -> Dict: """Send a direct message. Passing a user ID as the channel targets a DM.""" kwargs = {self._channel_param: user_id, self._text_param: text} return self.execute_tool(self._send_tool, **kwargs)

MCP connectors also wrap their payload. Every result arrives as a content envelope with a JSON string inside it, so the base connector unwraps before anything downstream sees it:

def _unwrap_mcp_envelope(data: Dict[str, Any]) -> Dict[str, Any]: """ MCP-based connectors (NOTIONMCP, SLACKMCP, AIRTABLEMCP, ...) return {"content": [{"type": "text", "text": ""}]} instead of a flat payload dict. Plain REST connectors (AIRTABLE, GOOGLEFORMS) return the flat payload directly, so this only unwraps when the envelope shape is actually present. """ if not isinstance(data, dict) or "content" not in data: return data content = data.get("content") if not isinstance(content, list) or not content: return data text = content[0].get("text") if isinstance(content[0], dict) else None if text is None: return data try: return json.loads(text) except (TypeError, ValueError): return {"text": text}

Skip the unwrap and data.get("results") returns None against a payload that clearly contains results, which reads exactly like an empty search.

Recommended reading: The posts on tool calling authentication for AI agents and credential ownership across agent tool-calling patterns go into when each surface is the right one.

Prerequisites

  • A Scalekit account; the free tier covers this. Credentials come from the dashboard under Developers, API Credentials.
  • An empty Airtable base. Just the base. The agent creates the table and fields for you, but Airtable's API has no create-base endpoint, so this one step stays manual.
  • A Google Form collecting free-text feedback, with a question identifying which employee the feedback is about. Create the questions by hand; see below for why.
  • A Notion workspace with a parent page to hold the per-employee summary pages, shared with your Scalekit integration.
  • A Slack workspace where the agent can DM the manager. No channel invite needed for a DM.
  • Python 3.11 or newer.
  • Optionally an OpenRouter API key. Read the next section before you set it.

The Data-Flow Decision You Should Make Before Running This

Setting OPENROUTER_API_KEY sends each employee's name and their raw feedback comments to a third-party API to generate the narrative summary. That is not a footnote on a performance review agent. It is the decision.

Leave it unset and summarization stays local. The rule-based path produces a deterministic summary from the same data: overall average, per-category averages, and the first five comments verbatim.

def summarize( feedback: EmployeeFeedback, review_period: str, openrouter_api_key: str, openrouter_model: str, ) -> str: """Summarize one employee's feedback. LLM first, rule-based fallback on any failure.""" if openrouter_api_key: try: return _summarize_with_llm(feedback, review_period, openrouter_api_key, openrouter_model) except Exception as e: logger.warning(f"LLM summarization failed for {feedback.name} ({e}), using rule-based summary") return _summarize_rule_based(feedback, review_period)

The tradeoff is real in both directions. The rule-based summary never leaves your connected services and never invents anything, but it is a table with comments underneath, not a narrative. The LLM summary reads like something a manager would write, and it costs you an egress path for employee feedback that your data-handling policy may not permit. Assess it before the first real cycle, not after.

The prompt itself is constrained accordingly: temperature is 0.3, and the last instruction is to not invent facts not present in the feedback.

_SUMMARY_PROMPT = """\ You are helping a manager write a fair, specific performance review summary. Employee: {name} Review period: {period} Average ratings by category: {ratings_block} Feedback comments collected from reviewers: {comments_block} Write a concise (4-6 sentence) narrative summary covering strengths, growth areas, and any patterns across reviewers. Be specific and reference the ratings where relevant. Do not invent facts not present in the feedback above. """

How to Set Up Your Connectors in Scalekit

Configure all four before running anything. Step 0 checks their status immediately, and having all four active means the first run exercises the whole path.

Step 1: Create Your Scalekit Account and Workspace

Create a free account, create a workspace for this project, and copy SCALEKIT_ENV_URL, SCALEKIT_CLIENT_ID, and SCALEKIT_CLIENT_SECRET into .env.

Step 2: Add the Airtable Connector

Go to AgentKit > Connections > Create Connection and add Airtable. Complete the OAuth flow and grant access to the base holding your review table. The provisioning step needs schema.bases:read to inspect the base and schema.bases:write to create the table; if you would rather create the table by hand, the read scope alone is enough.

Step 3: Add the Google Forms Connector

Add Google Forms with read access to your form and its responses.

Step 4: Add the Notion MCP Connector

Add a Notion MCP connection, not the plain Notion connector. Then share your parent page with the integration: open the page in Notion, click the three-dot menu, choose Connections, and add your Scalekit integration. Skipping the share step gives you an authorized connection that cannot see the page you configured.

Step 5: Add the Slack Connector

Add Slack with chat:write or the MCP equivalent. No channel invite is needed because the agent sends a direct message.

The single most common cause of a failed first run: Scalekit auto-suffixes connection names per workspace, so airtable comes back as airtable-3j16TKTG and notionmcp as notionmcp-chAb8Lfz. The Step 0 auth check calls get_or_create_connected_account() with whatever name you configured, and a generic provider label will not match. Copy the exact names from the dashboard.

# Connector names: copy the EXACT names from AgentKit > Connections. # Scalekit auto-suffixes these per workspace. AIRTABLE_CONNECTOR=airtable-3j16TKTG GOOGLE_FORMS_CONNECTOR=googleforms-WqF2XTWv NOTION_CONNECTOR=notionmcp-chAb8Lfz SLACK_CONNECTOR=slackmcp # Identity per connector: the identifier each connected account is keyed by AIRTABLE_USER=manager@yourcompany.com GOOGLE_FORMS_USER=manager@yourcompany.com NOTION_USER=manager@yourcompany.com SLACK_USER=manager@yourcompany.com # The manager this cycle is scoped to MANAGER_EMAIL=manager@yourcompany.com MANAGER_SLACK_ID=D01234567AB # DM conversation ID, or a C... channel ID # Sources and destination AIRTABLE_BASE_ID=appXXXXXXXXXXXXXX AIRTABLE_TABLE_NAME=Performance Reviews AIRTABLE_MANAGER_FIELD=Manager Email AIRTABLE_EMPLOYEE_FIELD=Employee Name GOOGLE_FORM_ID=1FAIpQLSc... FORM_EMPLOYEE_QUESTION_ID= NOTION_PARENT_PAGE_ID=... REVIEW_PERIOD=Q2 2026

Setting Up Auth with Claude Code

With the Scalekit plugin installed, the auth scaffold is two commands and a prompt.

claude plugin marketplace add scalekit-inc/claude-code-authstack claude plugin install agent-auth@scalekit-auth-stack

Then prompt Claude Code:

Set up Scalekit auth for Airtable, Google Forms, Notion MCP, and Slack MCP. I need to read records from an Airtable base, list responses from a Google Form, create and update pages in Notion, and DM a manager in Slack. Use actions.execute_tool() for all four, and unwrap the MCP content envelope on the Notion and Slack results.

It generates the client, the connector-to-identity map, and the shared base class every connector inherits from.

import scalekit.client sk = scalekit.client.ScalekitClient( client_id=cfg.scalekit_client_id, client_secret=cfg.scalekit_client_secret, env_url=cfg.scalekit_env_url, ) actions = sk.actions def get_connector_users(self) -> Dict[str, str]: """Mapping of connector name -> identifier, for auth checks.""" return { self.airtable_connector: self.airtable_user, self.google_forms_connector: self.google_forms_user, self.notion_connector: self.notion_user, self.slack_connector: self.slack_user, }

Four separate identity slots rather than one shared value is deliberate. In the common case they are all the manager's email. When they are not, that difference is exactly the thing you want visible in config instead of buried in a comment: an Airtable connection authorized by People Ops and a Slack connection authorized by the manager are two different principals, and the digest will read as coming from whoever holds the Slack grant.

The Base Connector

Every call in the pipeline goes through one method. identifier is what tells Scalekit which connected account to resolve.

class Connector: """Base connector class: shared auth-check and tool-execution logic.""" def __init__(self, actions, connector_name: str, identifier: str): self.actions = actions self.connector_name = connector_name self.identifier = identifier def check_auth(self) -> bool: """Check if connector is authorized. Returns True if ACTIVE.""" try: resp = self.actions.get_or_create_connected_account( connection_name=self.connector_name, identifier=self.identifier, ) status = resp.connected_account.status except Exception as e: logger.error(f"Failed to check {self.connector_name} auth: {e}") return False if status != "ACTIVE": logger.warning(f"{self.connector_name} ({self.identifier}) is {status}") try: link = self.actions.get_authorization_link( connection_name=self.connector_name, identifier=self.identifier, ).link logger.warning(f"Authorize here: {link}") except Exception: logger.warning("Check the Scalekit dashboard to authorize this connector") return False logger.info(f"{self.connector_name} ({self.identifier}) is ACTIVE") return True def execute_tool(self, tool_name: str, **kwargs) -> Dict[str, Any]: """Execute a Scalekit tool and return the data payload, unwrapping MCP envelopes.""" try: result = self.actions.execute_tool( tool_name=tool_name, identifier=self.identifier, tool_input=kwargs, ) return _unwrap_mcp_envelope(result.data or {}) except Exception as e: logger.error(f"Tool execution failed: {tool_name}: {e}") raise ConnectorError(f"{tool_name} failed: {e}") from e

The Pipeline, Step by Step

Step 0.5: Provision What Can Be Provisioned, Fail Loudly on What Cannot

Most agent tutorials assume the destination already exists. This one checks, creates what it can, and refuses to start against broken configuration.

def ensure_airtable_table( airtable: AirtableConnector, base_id: str, table_name: str, employee_field: str, manager_field: str, ) -> None: """ Ensure `table_name` exists in `base_id` with at least the employee/manager fields the agent requires. Creates the table with a default schema if it's missing. Raises ProvisioningError if the base itself doesn't exist or isn't accessible: Airtable's API cannot create a new base, only tables within one. """ try: schema = airtable.execute_tool("airtable_get_base_schema", base_id=base_id) or {} except ConnectorError as e: raise ProvisioningError( f"Cannot access Airtable base '{base_id}': {e}\n" f"Airtable's API cannot create a new base. Create an empty base at " f"airtable.com first, then set AIRTABLE_BASE_ID to its ID." ) from e existing_tables = {t.get("name") for t in schema.get("tables", [])} if table_name in existing_tables: logger.info(f"Airtable table '{table_name}' already exists") _ensure_required_fields(airtable, base_id, table_name, schema, employee_field, manager_field) return logger.warning(f"Airtable table '{table_name}' not found, creating it now") fields = list(_DEFAULT_FIELDS) field_names = {f["name"] for f in fields} if employee_field not in field_names: fields.insert(0, {"name": employee_field, "type": "singleLineText"}) if manager_field not in field_names: fields.insert(1, {"name": manager_field, "type": "singleLineText"}) try: airtable.execute_tool( "airtable_create_table", base_id=base_id, name=table_name, fields=fields, ) logger.info(f"Created Airtable table '{table_name}' with default review fields") except ConnectorError as e: raise ProvisioningError(f"Failed to create Airtable table '{table_name}': {e}") from e

The default schema is Employee Name, Manager Email, Communication Rating, Impact Rating, Comments. Note that airtable_create_table makes the first field in the array the primary field, which is why the employee field is inserted at index 0.

Google Forms gets validated but not provisioned, and the reason is a real connector limit rather than an oversight. The Google Forms connector exposes 10 tools, and googleforms_create_form takes only a title and an optional document title. There is no add-question tool. So the agent checks the form is reachable, warns if FORM_EMPLOYEE_QUESTION_ID matches nothing, and tells you to go add the questions by hand.

def validate_google_form(forms: GoogleFormsConnector, form_id: str, employee_question_id: str) -> None: try: form = forms.get_form(form_id) except ConnectorError as e: raise ProvisioningError( f"Cannot access Google Form '{form_id}': {e}\n" f"Create the form manually at forms.google.com with a question asking " f"'Which employee is this feedback about?', then set GOOGLE_FORM_ID." ) from e items = form.get("items", []) if not items: logger.warning( f"Google Form '{form_id}' has no questions yet. Google Forms cannot be " f"populated with questions via API, add them manually at forms.google.com." ) return if employee_question_id: question_ids = { item.get("questionItem", {}).get("question", {}).get("questionId") for item in items } if employee_question_id not in question_ids: logger.warning( f"FORM_EMPLOYEE_QUESTION_ID '{employee_question_id}' not found in form " f"'{form_id}', check googleforms_get_form output and update your .env" ) logger.info(f"Google Form '{form_id}' is accessible ({len(items)} question(s))")

Failing here exits with code 1 and an instruction, not a stack trace. An HR agent that silently proceeds against a form nobody has filled in produces empty summaries that look like poor performance.

Step 1: Paginate Both Sources to the End

Two sources, two pagination dialects. Airtable returns an offset token; Google Forms returns a nextPageToken. Both loops run to exhaustion, because a partial fetch in this pipeline means an employee's summary is missing feedback that exists.

def list_all_records( self, base_id: str, table_name: str, view: str = "", page_size: int = 100, ) -> List[Dict]: """Fetch every record in a table, paginating via Airtable's offset token.""" records: List[Dict] = [] offset = None while True: kwargs: Dict[str, Any] = { "base_id": base_id, "table_id_or_name": table_name, "page_size": page_size, } if view: kwargs["view"] = view if offset: kwargs["offset"] = offset data = self.execute_tool("airtable_list_records", **kwargs) or {} batch = data.get("records") or [] records.extend(batch) offset = data.get("offset") if not offset: break return records def list_all_responses(self, form_id: str, page_size: int = 100) -> List[Dict]: """Fetch every response submitted to a form, paginating via pageToken.""" responses: List[Dict] = [] page_token = None while True: kwargs: Dict[str, Any] = {"form_id": form_id, "page_size": page_size} if page_token: kwargs["page_token"] = page_token data = self.execute_tool("googleforms_list_responses", **kwargs) or {} batch = data.get("responses") or [] responses.extend(batch) page_token = data.get("nextPageToken") if not page_token: break return responses

Each source is fetched inside its own try, and a failure degrades to an empty list rather than aborting the cycle. If Airtable is down, the Form comments still make it into the summary, clearly labeled as having no ratings. Partial output beats no output here, because the alternative is a review cycle that stalls on an outage.

Step 2: Group by Employee, Tolerantly

Reviewers type names inconsistently. Matching is case-folded and whitespace-stripped, and bundles are keyed by the canonical spelling from direct_reports so the Notion page and the Slack digest agree.

bundles = {name: EmployeeFeedback(name) for name in direct_reports} normalized_to_canonical = {name.strip().casefold(): name for name in direct_reports} for record in airtable_records: fields = record.get("fields", {}) employee = fields.get(employee_field) canonical = normalized_to_canonical.get(str(employee).strip().casefold()) if employee else None if canonical: bundles[canonical].add_airtable_record(fields) for response in form_responses: answers = response.get("answers", {}) employee = _extract_form_employee(answers, form_employee_question_id) canonical = normalized_to_canonical.get(str(employee).strip().casefold()) if employee else None if not canonical: logger.warning( f"Form response references unrecognized employee {employee!r}, " f"skipping (does not match any direct report)" ) continue for question_id, answer in answers.items(): if question_id == form_employee_question_id: continue text = _extract_answer_text(answer) if text: bundles[canonical].add_form_comment(text)

Normalization handles "alex kim " against "Alex Kim". It does not handle a reviewer typing "Alex". That response is logged and dropped, which is the right call for review data (silently attaching feedback to the wrong person is worse than losing it) but it means the warning log is load-bearing. Read it after every cycle.

Rating columns are discovered by pattern rather than hardcoded, so the agent works across differently named review templates:

_RATING_FIELD_PATTERN = re.compile(r"(rating|score)", re.IGNORECASE)

Any Airtable field whose name matches, and whose value is numeric, is averaged. Non-numeric values in rating-looking columns are skipped rather than crashing the averaging.

Step 3: Upsert the Notion Page

Find by title, update if present, create if not.

def upsert_employee_page(self, parent_page_id: str, title: str, markdown_body: str) -> Dict: """Create the employee's page if it doesn't exist yet, otherwise update it in place.""" existing_id = self.find_existing_child_page(parent_page_id, title) if existing_id: logger.info(f"Updating existing Notion page for {title}") return self.update_employee_page(existing_id, markdown_body) logger.info(f"Creating new Notion page for {title}") return self.create_employee_page(parent_page_id, title, markdown_body) def create_employee_page(self, parent_page_id: str, title: str, markdown_body: str) -> Dict: """Create a new page under parent_page_id with the given title and markdown content.""" return self.execute_tool( "notionmcp_notion-create-pages", parent={"type": "page_id", "page_id": parent_page_id}, pages=[ { "properties": {"title": title}, "content": markdown_body, } ], ) def update_employee_page(self, page_id: str, markdown_body: str) -> Dict: """Overwrite an existing employee page's content with the latest summary.""" return self.execute_tool( "notionmcp_notion-update-page", page_id=page_id, command="replace_content", new_str=markdown_body, )

Two things about notionmcp_notion-search that matter more here than they would elsewhere. It is a semantic search across the entire workspace, not a scan of children under a parent page. And parent_page_id is accepted by find_existing_child_page() but never actually used to constrain the query; the match is on exact title alone. For a workspace with one review parent page that is fine. For a workspace where two teams each have an "Alex Kim, Q2 2026" page, the upsert can resolve to the wrong one. Add the parent page ID to the title, or scope the search with the tool's data_source_url parameter, before running this across multiple teams.

The page body is rendered as Notion-flavored markdown: a ratings table, the narrative, then the raw comments verbatim underneath. Keeping the raw feedback on the page is deliberate. A manager reading an LLM-written paragraph should be able to check it against the source without leaving the page.

Step 4: One DM Per Cycle

def build_slack_digest(manager_email: str, review_period: str, results: list) -> str: """Compose the manager's Slack DM summarizing what was written for each report.""" lines = [f"*Performance review summary ready, {review_period}*", ""] for entry in results: name = entry["name"] overall = entry["overall_average"] count = entry["response_count"] url = entry.get("notion_url", "") rating_str = f"{overall}/5 avg" if overall is not None else "no ratings" link = f" <{url}|View in Notion>" if url else "" lines.append(f"• *{name}*, {count} response(s), {rating_str}{link}") lines.append("") lines.append("_Summaries generated and written by your Performance Review Collector Agent._") return "\n".join(lines) digest = build_slack_digest(cfg.manager_email, cfg.review_period, results) slack.send_dm(cfg.manager_slack_id, digest) state.mark_processed(cfg.manager_email, cfg.review_period)

mark_processed() runs last, after the DM, and writes atomically via a temp file and rename. If the process dies before the DM lands, the cycle is not marked, and the next run retries. If it dies after, the manager is not re-pinged. Ordering that pair correctly is the difference between a cron job and a cron job that spams people.

Set MANAGER_SLACK_ID to a DM conversation ID (D...) or a channel ID (C...). It defaults to MANAGER_EMAIL, which only works if your Slack connector resolves DMs by email.

Running It

pip install -r requirements.txt cp .env.example .env # fill in your values python run_flow.py

One cycle, then exit. Schedule it weekly:

0 9 * * MON cd /path/to/performance-review-collector-agent && python run_flow.py

Or run it continuously during an active review window:

POLLING_MODE=true POLL_INTERVAL_MINUTES=60 python run_flow.py

Ctrl+C finishes the in-flight cycle and exits 130 without leaving partial Notion writes or a half-sent digest.

The exit codes are the monitoring surface, and they distinguish three states that all look like "nothing happened":

Code
Meaning
0
Summaries written, or no direct reports resolved (nothing to do)
1
Config missing, provisioning failed, or 5 consecutive polling errors
2
Direct reports were found, but none had any feedback this cycle
130
Graceful shutdown via Ctrl+C or SIGTERM

2 is the one to watch mid-cycle. It means reviewers have not submitted yet, which is a nudge-the-team signal rather than an engineering problem. Persistent 2 in the last week of a review window is worth an alert.

What to Check Before You Go Live

Decide the summarization path first. OPENROUTER_API_KEY set means employee feedback leaves your connected services. Unset means local, deterministic, and blunter. Make that call with whoever owns your data-handling policy, before the first real cycle.

Filter Airtable at the connector. Set AIRTABLE_VIEW to a per-manager view so rows outside the manager's scope never reach the process. It is one environment variable and it shrinks the blast radius immediately.

Set FORM_EMPLOYEE_QUESTION_ID explicitly. Leave it blank and the agent guesses the employee from the shortest text answer in each response, which is a heuristic that will eventually attribute someone's feedback to the wrong person. Get the real ID from googleforms_get_form.

Confirm the Notion parent is a page, not a database. A database ID produces a page that is created but empty, which reads like a content bug rather than a configuration one.

Fetch Forms responses incrementally once volume grows. googleforms_list_responses accepts a filter parameter on submission time, in the form timestamp > 2026-01-01T00:00:00Z. The template paginates everything every cycle; for a large form, filter to the current review window instead.

Log who read what. Four connected accounts touching employee review data is exactly the situation where "which credential read this row, on whose behalf, when" needs to be a query rather than an investigation. Audit trails for agent auth covers the event categories, and agent tool observability covers what to capture per tool call. Do this before HR asks, not after.

Delete the state file to reprocess. rm -f state/processed_cycles.json resets the manager-and-period guard. Useful for testing, dangerous on a shared box.

Conclusion

The retrieval problem is genuinely solved by four connectors and about two hundred lines. That part is not the interesting bit.

The interesting bit is that this agent reads feedback about people, and where you draw the credential boundary determines whether a mistake in your filtering logic is a bug or an incident. The template draws it in application code because that is what runs today on one manager's laptop. Moving it to the credential layer, one Airtable grant per manager, one connected account per identity, costs you a data model change and buys you the property that matters: what the manager cannot read, the agent cannot read.

The same pipeline extends without touching the auth layer. Swap Airtable for a BambooHR or Rippling connector and the ratings come from the HRIS. Add Google Docs and the summary lands in the review template your People Ops team already uses. Point Step 4 at a channel instead of a DM and it becomes a calibration prep digest. Once connected accounts are in place, adding a source is a new connector, not a new auth system.

The full code is on GitHub. Clone the repo, configure four connections, and run your first cycle in under 30 minutes. Start from the AgentKit quickstart, or browse the full connector catalog.

If you get stuck building your agent, join the Scalekit Slack community.

FAQ

Can I run this for every manager at once instead of one at a time?

The template scopes one cycle to one MANAGER_EMAIL. To fan out, loop over your manager list and pass each manager's real user ID as the identifier on every Scalekit call, rather than a single shared value from .env. That is also the change that makes the scoping real: each manager's connected account carries their own grant, so the isolation stops depending on your filter logic being right.

Why does it need the Notion MCP connector specifically?

Page-creation tools only exist on the MCP variant. The plain Notion connector exposes 51 tools to Notion MCP's 28 and still does not include notion-create-pages. If NOTION_CONNECTOR points at the plain connector, auth succeeds and Step 3 fails.

Why can the agent create the Airtable table but not the Google Form questions?

Because the connectors expose different surfaces. Airtable's API supports creating tables inside an existing base, so airtable_create_table handles it. The Google Forms connector's googleforms_create_form takes only a title; there is no add-question tool, so form structure stays a one-time manual step at forms.google.com. Airtable's base itself is also manual, for the same reason: no create-base endpoint exists.

What happens if one employee has no feedback yet?

They are skipped from both the Notion write and the Slack digest individually. Everyone else in the same cycle still gets processed. If nobody has feedback, the run exits 2 rather than writing empty pages.

Is re-running the agent safe?

Yes, in both directions. Notion pages are upserted by title, so a re-run updates in place instead of creating duplicates. The Slack DM is guarded by the processed-cycle state file, so the manager is not re-notified for a period already handled.

Should the agent use a shared HR service account or each manager's own credential?

Each manager's own credential, wherever the data model allows it. A shared account can read every employee's feedback, which makes every manager's digest correct only by convention and makes the audit trail useless (every read attributes to the bot). The full argument is in credential ownership in agent tool calling and why admin accounts are the wrong default for AI agents.

Do I have to manage token refresh across four connectors?

No. Tokens live in Scalekit's token vault and are refreshed before each tool call, which matters for a weekly cron that may not run for seven days between invocations. See how to handle token refresh for AI agents for why reactive refresh on a 401 is the wrong pattern for scheduled agents.

No items found.
Agent
Auth Quickstart
On this page
Share this article
Agent
Auth Quickstart

Acquire enterprise customers with
zero upfront cost.

Every feature unlocked. No hidden fees.