Airtable OAuth MCP Server

Instructions

Implementation Reference

• src/airtable_mcp/mcp/server.py:409-466 (handler)

  Primary handler for the 'search_records' MCP tool. Authenticates the client, normalizes the fields parameter to handle string or list input, constructs ListRecordsOptions, calls the AirtableClient.search_records helper, and returns formatted records with id, fields, and createdTime.

  @self.mcp.tool(description="Search records using a formula filter")
  async def search_records(
      base_id: Annotated[str, Field(description="The Airtable base ID")],
      table_id: Annotated[str, Field(description="The table ID or name")],
      filter_by_formula: Annotated[
          str, Field(description="Airtable formula for filtering")
      ],
      view: Annotated[str | None, Field(description="View name or ID")] = None,
      fields: Annotated[
          str | list[str] | None,
          Field(
              description="Specific fields to include (field name or array of field names)"
          ),
      ] = None,
  ) -> list[dict[str, Any]]:
      """Search records using a formula filter."""
      client = await self._get_authenticated_client()
  
      # Normalize fields parameter
      normalized_fields = None
      if fields is not None:
          if isinstance(fields, str):
              # Check if it's a JSON-encoded array string
              if fields.startswith("[") and fields.endswith("]"):
                  try:
                      import json
  
                      normalized_fields = json.loads(fields)
                  except (json.JSONDecodeError, ValueError):
                      # If JSON parsing fails, treat as single field name
                      normalized_fields = [fields]
              else:
                  # Single field name
                  normalized_fields = [fields]
          else:
              normalized_fields = fields
  
      options = ListRecordsOptions(
          view=view,
          fields=normalized_fields,
      )
  
      records = await client.search_records(
          base_id,
          table_id,
          filter_by_formula,
          options,
      )
  
      return [
          {
              "id": record.id,
              "fields": record.fields,
              "createdTime": record.created_time,
          }
          for record in records
      ]

• src/airtable_mcp/api/client.py:470-497 (helper)

  Supporting helper method on AirtableClient that applies the filter_by_formula to ListRecordsOptions and delegates to list_records method.

  async def search_records(
      self,
      base_id: str,
      table_id: str,
      filter_by_formula: str,
      options: ListRecordsOptions | None = None,
  ) -> list[AirtableRecord]:
      """Search records using a formula filter.
  
      Args:
          base_id: The Airtable base ID
          table_id: The table ID or name
          filter_by_formula: Airtable formula for filtering
          options: Additional options for the search
  
      Returns:
          List of matching records
      """
      logger.info(
          f"Searching records in {base_id}/{table_id} with formula: {filter_by_formula}"
      )
  
      # Merge filter with existing options
      search_options = options or ListRecordsOptions()
      search_options.filter_by_formula = filter_by_formula
  
      return await self.list_records(base_id, table_id, search_options)

• src/airtable_mcp/mcp/schemas.py:103-116 (schema)

  Pydantic schema model SearchRecordsArgs closely matching the search_records tool parameters, tested separately in unit tests.

  class SearchRecordsArgs(BaseArgs):
      """Arguments for search_records tool."""
  
      base_id: str = Field(description="The Airtable base ID")
      table_id: str = Field(description="The table ID or name")
      filter_by_formula: str = Field(description="Airtable formula for filtering")
      max_records: int | None = Field(
          default=None, description="Maximum number of records to return"
      )
      view: str | None = Field(default=None, description="View name or ID")
      fields: list[str] | None = Field(
          default=None, description="Specific fields to include"
      )