Polytomic MCP Server

Instructions

Implementation Reference

• src/polytomic_mcp/server.py:456-487 (handler)

  The create_bulk_sync function serves as the MCP tool handler for creating a new bulk sync (ELT) in Polytomic, parsing the input arguments and making a POST request to the Polytomic API.

  @mcp.tool()
  async def create_bulk_sync(
      name: str,
      source_connection_id: str,
      dest_connection_id: str,
      destination_configuration: str,
      schedule: str | None = None,
      source_configuration: str | None = None,
  ) -> str:
      """Create a new bulk sync (ELT) in Polytomic.
  
      Args:
          name: Name for the bulk sync
          source_connection_id: Source connection ID
          dest_connection_id: Destination connection ID
          destination_configuration: JSON string with destination config (e.g. {"schema": "public"})
          schedule: Optional JSON for schedule
          source_configuration: Optional JSON string with source-specific config
      """
      body = {
          "name": name,
          "source_connection_id": source_connection_id,
          "dest_connection_id": dest_connection_id,
          "destination_configuration": json.loads(destination_configuration),
      }
      if schedule:
          body["schedule"] = json.loads(schedule)
      if source_configuration:
          body["source_configuration"] = json.loads(source_configuration)
  
      result = await polytomic_request("/bulk/syncs", method="POST", body=body)
      return json.dumps(result, indent=2)